1.什么是生成者与消费者模型
生产者-消费者模型是并发编程中的经典问题,描述了多个线程(或进程)如何安全、有效地共享有限的缓冲区资源。在这个模型中,有两种角色:
-
生产者(Producer):负责生成数据或者将数据放置到共享缓冲区中。
-
消费者(Consumer):负责从共享缓冲区中取出数据并进行处理或消费。
核心概念:
生产者和消费者模型主要涉及以下几个关键概念:
-
共享缓冲区(Shared Buffer):生产者和消费者之间共享的有限大小的缓冲区。这个缓冲区可以是一个队列或者一个固定大小的数组。
-
同步(Synchronization):确保生产者和消费者之间的正确协作,避免数据竞争和资源争用。例如,当缓冲区已满时,生产者应该等待;当缓冲区为空时,消费者应该等待。
-
互斥(Mutual Exclusion):确保同一时刻只有一个线程(生产者或消费者)可以访问或操作共享缓冲区,以避免数据不一致或丢失。
具体案例:
假设有一个生产者和多个消费者的情形,以一个有界缓冲区(Bounded Buffer)为例:
-
共享缓冲区:假设有一个大小为10的数组作为共享缓冲区,用于存放生产者生产的产品。
-
生产者:负责生成产品,并将产品放入共享缓冲区中。如果缓冲区已满,生产者需要等待,直到有空间可以放置产品。
-
消费者:负责从共享缓冲区中取出产品并进行消费。如果缓冲区为空,消费者需要等待,直到有产品可以消费。
解决方案:
为了解决生产者-消费者模型中的同步和互斥问题,可以采用以下方法:
-
互斥锁(Mutex):确保在同一时刻只有一个线程可以访问或修改共享缓冲区。例如,生产者和消费者在访问缓冲区前,首先要获取互斥锁,操作完成后释放锁。
-
条件变量(Condition Variables):用于线程间的通信,如通知生产者缓冲区有空间可以放置新产品,或通知消费者缓冲区中有产品可以消费。
-
信号量(Semaphores):用于控制对共享资源的访问,如控制缓冲区的空闲空间数量或产品数量。
示例代码:
使用C语言来实现,包括两个生产者线程、三个消费者线程,一个大小为10的共享缓冲区(使用链表实现),每个线程生成或消费一个数据后休眠1-2秒,并打印过程。我们将使用条件变量来实现线程的等待和唤醒机制。
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#define BUFFER_SIZE 10
// 链表节点
typedef struct Node {
int data;
struct Node *next;
} Node;
// 全局变量
Node *head = NULL; // 指向链表头部的指针
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 互斥锁
pthread_cond_t empty = PTHREAD_COND_INITIALIZER; // 缓冲区为空的条件变量
pthread_cond_t full = PTHREAD_COND_INITIALIZER; // 缓冲区已满的条件变量
int count = 0; // 缓冲区当前数据项的数量
// 生产者线程函数
void *producer_func(void *arg) {
int id = *((int *)arg); // 生产者线程的编号
int data = 0;
while (1) {
// 生成数据项
data++;
// 获取互斥锁
pthread_mutex_lock(&mutex);
// 等待直到缓冲区有空间
while (count == BUFFER_SIZE) {
printf("Producer %d: Buffer is full. Waiting...\n", id);
//条件变量 (empty):在缓冲区已满时,生产者线程会等待在 empty 条件变量上,直到有消费者取走数据并通知它。
pthread_cond_wait(&empty, &mutex);
}
// 将数据项放入缓冲区,头插法,取节点的时候不用遍历链表
Node *new_node = (Node *)malloc(sizeof(Node));
new_node->data = data;
new_node->next = head;
head = new_node;
count++;
printf("Producer %d: Produced data: %d\n", id, data);
// 在生产者放入数据后,会通过 full 条件变量唤醒等待的消费者线程,告知它们可以消费数据了。
pthread_cond_signal(&full);
// 释放互斥锁
pthread_mutex_unlock(&mutex);
// 休眠1-2秒
sleep(rand() % 2);
}
pthread_exit(NULL);
}
// 消费者线程函数
void *consumer_func(void *arg) {
int id = *((int *)arg); // 消费者线程的编号
while (1) {
// 获取互斥锁
pthread_mutex_lock(&mutex);
// 等待直到缓冲区有数据
while (head == NULL) {
printf("Consumer %d: Buffer is empty. Waiting...\n", id);
//条件变量 (full):在缓冲区为空时,消费者线程会等待在 full 条件变量上,直到有生产者放入数据并通知它。
pthread_cond_wait(&full, &mutex);
}
// 从缓冲区取出数据项
Node *temp = head;
head = head->next;
int data = temp->data;
free(temp);
count--;
printf("Consumer %d: Consumed data: %d\n", id, data);
// 在消费者取走数据后,会通过 empty 条件变量唤醒等待的生产者线程,告知它们可以继续生产数据。
pthread_cond_signal(&empty);
// 释放互斥锁
pthread_mutex_unlock(&mutex);
// 休眠1-3秒
sleep(rand() % 3 + 1);
}
pthread_exit(NULL);
}
int main() {
pthread_t producers[2], consumers[3];
int producer_ids[2] = {1, 2}; // 两个生产者线程的编号
int consumer_ids[3] = {1, 2, 3}; // 三个消费者线程的编号
// 初始化互斥锁和条件变量
pthread_mutex_init(&mutex, NULL);
pthread_cond_init(&empty, NULL);
pthread_cond_init(&full, NULL);
// 创建生产者线程
for (int i = 0; i < 2; ++i) {
pthread_create(&producers[i], NULL, producer_func, (void *)&producer_ids[i]);
}
// 创建消费者线程
for (int i = 0; i < 3; ++i) {
pthread_create(&consumers[i], NULL, consumer_func, (void *)&consumer_ids[i]);
}
// 主线程等待所有子线程结束
for (int i = 0; i < 2; ++i) {
pthread_join(producers[i], NULL);
}
for (int i = 0; i < 3; ++i) {
pthread_join(consumers[i], NULL);
}
// 销毁互斥锁和条件变量
pthread_mutex_destroy(&mutex);
pthread_cond_destroy(&empty);
pthread_cond_destroy(&full);
return 0;
}