Linux多线程编程:用互斥锁与条件变量解决多生产者-多消费者问题

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

#include <unistd.h>

#include <time.h>

#define BUFFER_SIZE 5  // 缓冲区大小

#define MAX_NUM 20     // 总生成数

#define PRODUCERS 2    // 生产者数量

#define CONSUMERS 3    // 消费者数量

#define PRODUCER_NUM (MAX_NUM / PRODUCERS)  // 每个生产者生成数

// 缓冲区及控制变量

int buffer[BUFFER_SIZE];

int in = 0;          // 生产者放入位置

int out = 0;         // 消费者取出位置

int count = 0;       // 缓冲区内数据个数

int total_produced = 0;  // 已生成总数

// 同步机制

pthread_mutex_t mutex;

pthread_cond_t not_empty;  // 缓冲区非空条件

pthread_cond_t not_full;   // 缓冲区非满条件

// 生产者线程函数

void *producer(void *arg) {

   int id = *(int *)arg;

   free(arg);  // 释放动态分配的ID

   for (int i = 0; i < PRODUCER_NUM; i++) {

       // 生成1-MAX_NUM的随机数

       int num = rand() % MAX_NUM + 1;

       usleep(rand() % 500000);  // 随机延时模拟生产耗时

       pthread_mutex_lock(&mutex);

       // 若缓冲区满，等待非满条件

       while (count == BUFFER_SIZE) {

           pthread_cond_wait(&not_full, &mutex);

       }

       // 放入数据

       buffer[in] = num;

       printf("生产者%d：放入 %d（位置：%d，当前数量：%d）\n", id, num, in, count + 1);

       in = (in + 1) % BUFFER_SIZE;

       count++;

       total_produced++;

       pthread_cond_signal(&not_empty);  // 唤醒等待非空的消费者

       pthread_mutex_unlock(&mutex);

   }

   printf("生产者%d：完成生产，退出\n", id);

   pthread_exit(NULL);

}

// 消费者线程函数

void *consumer(void *arg) {

   int id = *(int *)arg;

   free(arg);  // 释放动态分配的ID

   while (1) {

       pthread_mutex_lock(&mutex);

       // 若缓冲区空且生产完毕，退出

       while (count == 0 && total_produced >= MAX_NUM) {

           pthread_mutex_unlock(&mutex);

           printf("消费者%d：无数据可消费，退出\n", id);

           pthread_exit(NULL);

       }

       // 若缓冲区空，等待非空条件

       while (count == 0) {

           pthread_cond_wait(&not_empty, &mutex);

       }

       // 取出数据

       int num = buffer[out];

       printf("消费者%d：取出 %d（位置：%d，当前数量：%d）\n", id, num, out, count - 1);

       out = (out + 1) % BUFFER_SIZE;

       count--;

       pthread_cond_signal(&not_full);  // 唤醒等待非满的生产者

       pthread_mutex_unlock(&mutex);

       usleep(rand() % 500000);  // 随机延时模拟消费耗时

   }

}

int main() {

   pthread_t prod_tids[PRODUCERS];

   pthread_t cons_tids[CONSUMERS];

   srand(time(NULL));  // 初始化随机数种子

   // 初始化同步机制

   pthread_mutex_init(&mutex, NULL);

   pthread_cond_init(&not_empty, NULL);

   pthread_cond_init(&not_full, NULL);

   // 创建生产者线程

   for (int i = 0; i < PRODUCERS; i++) {

       int *id = malloc(sizeof(int));

       *id = i + 1;  // 生产者ID：1,2

       if (pthread_create(&prod_tids[i], NULL, producer, id) != 0) {

           perror("pthread_create producer failed");

           exit(EXIT_FAILURE);

       }

   }

   // 创建消费者线程

   for (int i = 0; i < CONSUMERS; i++) {

       int *id = malloc(sizeof(int));

       *id = i + 1;  // 消费者ID：1,2,3

       if (pthread_create(&cons_tids[i], NULL, consumer, id) != 0) {

           perror("pthread_create consumer failed");

           exit(EXIT_FAILURE);

       }

   }

   // 等待所有生产者线程结束

   for (int i = 0; i < PRODUCERS; i++) {

       pthread_join(prod_tids[i], NULL);

   }

   // 唤醒所有等待的消费者（确保它们检查退出条件）

   pthread_mutex_lock(&mutex);

   pthread_cond_broadcast(&not_empty);

   pthread_mutex_unlock(&mutex);

   // 等待所有消费者线程结束

   for (int i = 0; i < CONSUMERS; i++) {

       pthread_join(cons_tids[i], NULL);

   }

   // 清理资源

   pthread_mutex_destroy(&mutex);

   pthread_cond_destroy(&not_empty);

   pthread_cond_destroy(&not_full);

   printf("所有线程结束，程序退出\n");

   return 0;

}

生产者1：放入 5（位置：0，当前数量：1）

消费者1：取出 5（位置：0，当前数量：0）

生产者2：放入 12（位置：1，当前数量：1）

生产者1：放入 3（位置：0，当前数量：2）

消费者2：取出 12（位置：1，当前数量：1）

...

生产者2：完成生产，退出

生产者1：完成生产，退出

消费者3：取出 18（位置：3，当前数量：0）

消费者1：无数据可消费，退出

消费者2：无数据可消费，退出

消费者3：无数据可消费，退出

所有线程结束，程序退出