Linux多线程编程-生产者与消费者模型详解与实现（C语言） ...

1.什么是生成者与消费者模型

生产者-消费者模型是并发编程中的经典问题，描述了多个线程（或历程）怎样安全、有效地共享有限的缓冲区资源。在这个模型中，有两种角色：

• 生产者（Producer）：负责生成数据或者将数据放置到共享缓冲区中。
  
• 消费者（Consumer）：负责从共享缓冲区中取出数据并进行处理或消费。
  

核心概念：

生产者和消费者模型主要涉及以下几个关键概念：

• 共享缓冲区（Shared Buffer）：生产者和消费者之间共享的有限大小的缓冲区。这个缓冲区可以是一个队列或者一个固定大小的数组。
  
• 同步（Synchronization）：确保生产者和消费者之间的正确协作，制止数据竞争和资源争用。例如，当缓冲区已满时，生产者应该等候；当缓冲区为空时，消费者应该等候。
  
• 互斥（Mutual Exclusion）：确保同一时候只有一个线程（生产者或消费者）可以访问或操纵共享缓冲区，以制止数据不一致或丢失。
  

详细案例：

假设有一个生产者和多个消费者的情况，以一个有界缓冲区（Bounded Buffer）为例：

• 共享缓冲区：假设有一个大小为10的数组作为共享缓冲区，用于存放生产者生产的产物。
  
• 生产者：负责生成产物，并将产物放入共享缓冲区中。如果缓冲区已满，生产者需要等候，直到有空间可以放置产物。
  
• 消费者：负责从共享缓冲区中取出产物并进行消费。如果缓冲区为空，消费者需要等候，直到有产物可以消费。
  

解决方案：

为相识决生产者-消费者模型中的同步和互斥问题，可以采用以下方法：

• 互斥锁（Mutex）：确保在同一时候只有一个线程可以访问或修改共享缓冲区。例如，生产者和消费者在访问缓冲区前，首先要获取互斥锁，操纵完成后开释锁。
  
• 条件变量（Condition Variables）：用于线程间的通讯，如关照生产者缓冲区有空间可以放置新产物，或关照消费者缓冲区中有产物可以消费。
  
• 信号量（Semaphores）：用于控制对共享资源的访问，如控制缓冲区的空闲空间数量或产物数量。
  

示例代码：

使用C语言来实现，包括两个生产者线程、三个消费者线程，一个大小为10的共享缓冲区（使用链表实现），每个线程生成或消费一个数据后休眠1-2秒，并打印过程。我们将使用条件变量来实现线程的等候和唤醒机制。

1. #include <stdio.h>
2. #include <stdlib.h>
3. #include <pthread.h>
4. #include <unistd.h>
6. #define BUFFER_SIZE 10
8. // 链表节点
9. typedef struct Node {
10.     int data;
11.     struct Node *next;
12. } Node;
14. // 全局变量
15. Node *head = NULL;                  // 指向链表头部的指针
16. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  // 互斥锁
17. pthread_cond_t empty = PTHREAD_COND_INITIALIZER;    // 缓冲区为空的条件变量
18. pthread_cond_t full = PTHREAD_COND_INITIALIZER;     // 缓冲区已满的条件变量
19. int count = 0;                      // 缓冲区当前数据项的数量
21. // 生产者线程函数
22. void *producer_func(void *arg) {
23.     int id = *((int *)arg);  // 生产者线程的编号
24.     int data = 0;
25.    
26.     while (1) {
27.         // 生成数据项
28.         data++;
29.         
30.         // 获取互斥锁
31.         pthread_mutex_lock(&mutex);
32.         
33.         // 等待直到缓冲区有空间
34.         while (count == BUFFER_SIZE) {
35.             printf("Producer %d: Buffer is full. Waiting...\n", id);
36.             //条件变量 (empty)：在缓冲区已满时，生产者线程会等待在 empty 条件变量上，直到有消费者取走数据并通知它。
37.             pthread_cond_wait(&empty, &mutex);
38.         }
39.         
40.         // 将数据项放入缓冲区,头插法，取节点的时候不用遍历链表
41.         Node *new_node = (Node *)malloc(sizeof(Node));
42.         new_node->data = data;
43.         new_node->next = head;
44.         head = new_node;
45.         count++;
46.         
47.         printf("Producer %d: Produced data: %d\n", id, data);
48.         
49.         // 在生产者放入数据后，会通过 full 条件变量唤醒等待的消费者线程，告知它们可以消费数据了。
50.         pthread_cond_signal(&full);
51.         
52.         // 释放互斥锁
53.         pthread_mutex_unlock(&mutex);
54.         
55.         // 休眠1-2秒
56.         sleep(rand() % 2);
57.     }
58.    
59.     pthread_exit(NULL);
60. }
62. // 消费者线程函数
63. void *consumer_func(void *arg) {
64.     int id = *((int *)arg);  // 消费者线程的编号
65.    
66.     while (1) {
67.         // 获取互斥锁
68.         pthread_mutex_lock(&mutex);
69.         
70.         // 等待直到缓冲区有数据
71.         while (head == NULL) {
72.             printf("Consumer %d: Buffer is empty. Waiting...\n", id);
73.             //条件变量 (full)：在缓冲区为空时，消费者线程会等待在 full 条件变量上，直到有生产者放入数据并通知它。
74.             pthread_cond_wait(&full, &mutex);
75.         }
76.         
77.         // 从缓冲区取出数据项
78.         Node *temp = head;
79.         head = head->next;
80.         int data = temp->data;
81.         free(temp);
82.         count--;
83.         
84.         printf("Consumer %d: Consumed data: %d\n", id, data);
85.         
86.         // 在消费者取走数据后，会通过 empty 条件变量唤醒等待的生产者线程，告知它们可以继续生产数据。
87.         pthread_cond_signal(&empty);
88.         
89.         // 释放互斥锁
90.         pthread_mutex_unlock(&mutex);
91.         
92.         // 休眠1-3秒
93.         sleep(rand() % 3 + 1);
94.     }
95.    
96.     pthread_exit(NULL);
97. }
99. int main() {
100.     pthread_t producers[2], consumers[3];
101.     int producer_ids[2] = {1, 2};  // 两个生产者线程的编号
102.     int consumer_ids[3] = {1, 2, 3};  // 三个消费者线程的编号
103.    
104.     // 初始化互斥锁和条件变量
105.     pthread_mutex_init(&mutex, NULL);
106.     pthread_cond_init(&empty, NULL);
107.     pthread_cond_init(&full, NULL);
108.    
109.     // 创建生产者线程
110.     for (int i = 0; i < 2; ++i) {
111.         pthread_create(&producers[i], NULL, producer_func, (void *)&producer_ids[i]);
112.     }
113.    
114.     // 创建消费者线程
115.     for (int i = 0; i < 3; ++i) {
116.         pthread_create(&consumers[i], NULL, consumer_func, (void *)&consumer_ids[i]);
117.     }
118.    
119.     // 主线程等待所有子线程结束
120.     for (int i = 0; i < 2; ++i) {
121.         pthread_join(producers[i], NULL);
122.     }
123.     for (int i = 0; i < 3; ++i) {
124.         pthread_join(consumers[i], NULL);
125.     }
126.    
127.     // 销毁互斥锁和条件变量
128.     pthread_mutex_destroy(&mutex);
129.     pthread_cond_destroy(&empty);
130.     pthread_cond_destroy(&full);
131.    
132.     return 0;
133. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。