编写生产者和消费者程序
编写生产者和消费者程序，要求：
- 生产者和消费者两个程序，共用一个仓库，仓库是一个普通文件（/tmp/store），容量为100个字节；
- 生产者生产资源放进仓库，消费者则从仓库中消费资源；资源为数字字符“0-9”，一个资源就是一个数字，10个数字循环生成；
- 生产者创建仓库（/tmp/store），间隔1s生产一个资源，当仓库满了（资源数量达到100个）的时候，生产者不能继续生产；消费者间隔2s消费一个资源，当仓库为空的时候，消费者不能继续消费；
- 消费者每次消费1个资源，首先打印出消耗之前仓库中的资源数量和空位的数量，然后打印出消耗之后仓库中的资源数量和空位的数量，并打印出所消耗的资源内容；
- 生产者每次生产1个资源，先打印出生产之前仓库中的资源数量和空位的数量，然后打印出生产之后仓库中的资源数量和空位的数量，并打印出所生产的资源内容。
- 消费者消费资源后需要把已经消费的资源从仓库里删除；
- 用信号量实现进程的同步和互斥。
【提示】题目有多种解决方案，可以用1个或多个信号量，或者使用其他合适的方法。
- 请看👉 ：用管程解决生产者消费者问题
这是我之前写的一个，可以看看
public class MyProducer {
public static void main(String[] args) {
/*Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.153.139:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
*//**
* 0 producer在发送数据后，不会等待broker任何响应，无法确保数据正确送到broker中
* 1 producer在发送数据后，只需要得到副本（leader,follower）中leader确认就可以，
* kafka默认方式，有可能会丢失数据，数据不会重复
* -1（all） producer需要等到所有的副本全部确认，响应时间最长，数据最安全，
* 不会丢失数据，但数据可能重复
*//*
properties.put(ProducerConfig.ACKS_CONFIG,"-1");
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
Scanner scanner = new Scanner(System.in);
String tag = "1";
while(tag.equals("1")){
System.out.print("请输出要发送到Kafka的内容：");
String content = scanner.nextLine();
ProducerRecord<String, String> record = new ProducerRecord<>("mydemo2", content);
producer.send(record);
System.out.print("是否退出？ 0：退出 1:继续发送 请输入：");
tag = scanner.nextLine();
}*/
// 多线程
long start = System.currentTimeMillis();
for (int i=0; i<1000; i++){
new Thread(new Runnable() {
@Override
public void run() {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.153.139:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
properties.put(ProducerConfig.ACKS_CONFIG,"-1");
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
for (int j=0; j<100000;j++){
String name = Thread.currentThread().getName();
String sendMsg = "currentThread name is "+name + " send msg count " + j;
ProducerRecord<String, String> record = new ProducerRecord<>("mydemo2",
sendMsg);
producer.send(record);
System.out.println(sendMsg);
if (j%1000==0){
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}).start();
}
System.out.println("time: "+ (System.currentTimeMillis()-start) );
System.out.println("game over");
}
}
public class MyConsumer {
public static void main(String[] args) {
final Properties prop = new Properties();
prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.153.139:9092");
prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "10000");
// earliest 当各分区下有已经提交的offset ,从上次提交的offset后开始消费
// 如果没有查到已经提交的offset, 从分区内的第一条消息开始消费
// latest 当各分区下有已经提交的offset ,从上次提交的offset后开始消费
// 如果没有查到已经提交的offset,消费新产生的分区下的消息
// none 当各分区下有已经提交的offset ,从上次提交的offset后开始消费
// 如果没有查到已经提交的offset,抛出异常
prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
// 每一个消费，都要定义不同的Group_ID
prop.put(ConsumerConfig.GROUP_ID_CONFIG,"group_5");
/* 一个消费都组，只有一个消费者
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(prop);
kafkaConsumer.subscribe(Collections.singleton("mydemo"));
while (true){
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
for (ConsumerRecord<String,String> record: records) {
System.out.println(record.offset()+" "+ record.key()+" "+record.value());
System.out.println();
}
System.out.println("-------------------------------");
// 如果自动提交关闭 ENABLE_AUTO_COMMIT_CONFIG设置为"false"
// 手动提交方法
// kafkaConsumer.commitAsync();
}*/
for (int i = 0; i < 12; i++) {
new Thread(new Runnable() {
@Override
public void run() {
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(prop);
kafkaConsumer.subscribe(Collections.singleton("mydemo2"));
while (true){
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
String content = Thread.currentThread().getName() + " 没有数据";
for (ConsumerRecord<String,String> record: records) {
content = Thread.currentThread().getName()+" "+ record.offset()+" "
+ record.key()+" "+record.value();
System.out.println(content);
}
if (records.isEmpty()){
System.out.println(content);
}
}
}
}).start();
}
}
}
用信号量实现同步和互斥是最容易的了，这种需求基本上所有语言都支持的
新代码运行结果
生产者：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/sem.h>
int semid;
union semun
{
int val;
};
//初始化
void sem_init()
{
//semget：创建一个新信号量或者获取一个已有的信号量的键
semid=semget((key_t)1234,1,IPC_CREAT|IPC_EXCL|0600);//1234:房间号,0600->文件权限
if(semid==-1) //全新创建失败,获取已经存在的信号量
{
semid=semget((key_t)1234,1,0600);
if(semid==-1)
{
perror("semget error");
}
else
{
//初始化
union semun a;//定义联合体
a.val=1;//初始化设置为1，假设刚开始可以使用
//semctl：用来直接控制信号量信息
if(semctl(semid,0,SETVAL,a)==-1)//编号从0开始
{
perror("semctl init error");
}
}
}
}
void sem_p()//p操作
{
struct sembuf buf;
buf.sem_num=0;
buf.sem_op=-1;
buf.sem_flg=SEM_UNDO;//一般都这样设置，是让操作系统记住使用了P操作
//semop：对信号量进行改变，做p或者v操作
if(semop(semid,&buf,1)==-1)//1:长度，只有一个信号量
perror("p perror");
}
void sem_v()//v操作
{
struct sembuf buf;
buf.sem_num=0;
buf.sem_op=1;
buf.sem_flg=SEM_UNDO;//一般都这样设置，是让操作系统记住使用了P操作
if(semop(semid,&buf,1)==-1)//1:长度，只有一个信号量
perror("v perror");
}
//销毁
void sem_destory()
{
if(semctl(semid,0,IPC_RMID)==-1)
perror("destory sem error");
}
int main()
{
int i=0;
int size=0;
FILE* fp;
//char sym[]={'0','1','2','3','4','5','6','7','8','9'};
char ch;
//信号量初始化
sem_init();
printf("This is the producer...\n");
while(1)
{
sem_p();//
fp = fopen("/tmp/store","a");
size = ftell(fp); //
printf("Before produce, resource number:%d,left position:%d\n",size,100-size);
if(size >=100)
{
printf("The store is full! resource number:%d,left position:%d\n",size,100-size);
fclose(fp);
}else
{
ch = (char)('0'+i%10);
printf("After produce, resource number:%d,left position:%d, ",size+1,99-size);
printf("new resource:%c\n",ch);
i++;
fwrite(&ch,1,1,fp);
fclose(fp);
}
sem_v();
sleep(1);
}
sem_destory();
}
消费者：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/sem.h>
int semid;
union semun
{
int val;
};
//初始化
void sem_init()
{
//semget：创建一个新信号量或者获取一个已有的信号量的键
semid=semget((key_t)1234,1,IPC_CREAT|IPC_EXCL|0600);//1234:房间号,0600->文件权限
if(semid==-1) //全新创建失败,获取已经存在的信号量
{
semid=semget((key_t)1234,1,0600);
if(semid==-1)
{
perror("semget error");
}
else
{
//初始化
union semun a;//定义联合体
a.val=1;//初始化设置为1，假设刚开始可以使用
//semctl：用来直接控制信号量信息
if(semctl(semid,0,SETVAL,a)==-1)//编号从0开始
{
perror("semctl init error");
}
}
}
}
void sem_p()//p操作
{
struct sembuf buf;
buf.sem_num=0;
buf.sem_op=-1;
buf.sem_flg=SEM_UNDO;//一般都这样设置，是让操作系统记住使用了P操作
//semop：对信号量进行改变，做p或者v操作
if(semop(semid,&buf,1)==-1)//1:长度，只有一个信号量
perror("p perror");
}
void sem_v()//v操作
{
struct sembuf buf;
buf.sem_num=0;
buf.sem_op=1;
buf.sem_flg=SEM_UNDO;//一般都这样设置，是让操作系统记住使用了P操作
if(semop(semid,&buf,1)==-1)//1:长度，只有一个信号量
perror("v perror");
}
//销毁
void sem_destory()
{
if(semctl(semid,0,IPC_RMID)==-1)
perror("destory sem error");
}
int main()
{
int size=0;
FILE* fp,*fp2;
char ch[110]={0};
int len,i;
//信号量初始化
sem_init();
printf("This is a customer...\n");
while(1)
{
sem_p();//
memset(ch,0,110);
fp = fopen("/tmp/store","r");
fseek(fp,0,SEEK_END);
size = ftell(fp); //
fseek(fp,0,SEEK_SET);
fread(ch,1,size,fp);
fclose(fp);
/*for(i=0;i<size;i++)
printf("%c ",ch[i]);
printf("\n");*/
len = strlen(ch);
ch[len] = 0;
printf("Before use, resource number:%d, left position:%d\n",len,100-len);
if(len == 0)
{
printf("The store is empty!\n");
}
else
{
printf("After use, resource number:%d, left position:%d, ",len-1,101-len);
printf("the resource:%c\n",ch[0]);
fp2 = fopen("/tmp/store","w");
fwrite(ch+1,1,len-1,fp2);
fclose(fp2);
}
sem_v();
sleep(2);
}
sem_destory();
}
这里提供2篇不错的Java实现方式，供你参考
链接1：https://blog.csdn.net/m0_63694520/article/details/125905458
链接2：https://blog.csdn.net/zhengzhaoyang122/article/details/115803193