（十）生产者-消费者模型
生产者-消费者模型是一种经典的并发编程模型，旨在解决多线程环境中负责生成数据或任务的线程（生产者）与负责处理数据或任务的线程（消费者）之间的协作与数据共享问题。生产者-消费者模型的关键要素包括：
-
共享缓冲区：用于存放由生产者生产，被消费者处理的数据或任务。这个缓冲区通常是一个队列。
-
同步机制：用于确保生产者和消费者线程之间的有效协调和数据安全共享。
单消费者-单生产者示例
下面的示例实现了一个单生产者-单消费者模型，以 std::queue 作为共享缓冲区，并通过 std::mutex 和 std::condition_variable 实现线程的同步。
#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
std::queue<int> buffer; // 生产者消费者共享的缓冲区，用来存放数据或任务
const int max_buffer_size = 10; // 缓冲区最大容量
std::mutex mtx; // 保护共享资源的互斥锁，确保在同一时间只有一个线程可以访问缓冲区
std::condition_variable cv_producer, cv_consumer; // 生产者和消费者的条件变量
bool finished = false; // 标记生产者是否已经完成
void producer() {
for (int i = 1; i <= 20; ++i) {
std::unique_lock<std::mutex> lock(mtx); // 获取互斥锁
cv_producer.wait(lock, []{ return buffer.size() < max_buffer_size; }); // 等待直到缓冲区有空间
buffer.push(i); // 生产一个数据并放入缓冲区
std::cout << "Produced: " << i << std::endl;
cv_consumer.notify_one(); // 通知消费者有新数据可用
}
std::unique_lock<std::mutex> lock(mtx);
finished = true; // 设置生产者已完成标志
cv_consumer.notify_all(); // 通知所有消费者
}
void consumer() {
while (true) {
std::unique_lock<std::mutex> lock(mtx); // 获取互斥锁
cv_consumer.wait(lock, []{ return !buffer.empty() || finished; }); // 等待直到缓冲区不为空或生产者已完成
if (buffer.empty() && finished) {
break; // 如果缓冲区为空且生产者已完成，退出循环
}
int item = buffer.front(); // 从缓冲区消费一个数据
buffer.pop(); // 移除已消费的数据
std::cout << "Consumed: " << item << std::endl;
cv_producer.notify_one(); // 通知生产者缓冲区有空间
}
}
int main() {
std::thread producer_thread(producer);
std::thread consumer_thread(consumer);
producer_thread.join();
consumer_thread.join();
return 0;
}
Produced: 1
Produced: 2
Produced: 3
Produced: 4
Produced: 5
Produced: 6
Produced: 7
Produced: 8
Produced: 9
Produced: 10
Consumed: 1
Consumed: 2
Consumed: 3
Consumed: 4
Consumed: 5
Consumed: 6
Consumed: 7
Consumed: 8
Consumed: 9
Consumed: 10
Produced: 11
Produced: 12
Produced: 13
Produced: 14
Produced: 15
Produced: 16
Produced: 17
Produced: 18
Produced: 19
Produced: 20
Consumed: 11
Consumed: 12
Consumed: 13
Consumed: 14
Consumed: 15
Consumed: 16
Consumed: 17
Consumed: 18
Consumed: 19
Consumed: 20
多生产者-多消费者示例
在实际应用中，通常会有多个消费者来处理多个生产者生成的数据。这种设置可以提高系统的并发性和吞吐量。以下是一个多生产者-多消费者模型的示例：
#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <vector>
const int max_buffer_size = 10;
std::queue<int> buffer;
std::mutex mtx;
std::condition_variable cv_producer, cv_consumer;
int next_item = 1; // 共享的生产数据计数器
bool should_quit = false; // 共享的退出标志
void producer(int id) {
while (true) {
std::unique_lock<std::mutex> lock(mtx); // 获取互斥锁
cv_producer.wait(lock, []{ return buffer.size() < max_buffer_size || should_quit; }); // 等待直到缓冲区有空间或需要退出
if (should_quit) break; // 如果需要退出，则退出生产者线程
buffer.push(next_item); // 生产一个数据并放入缓冲区
std::cout << "Producer " << id << " produced: " << next_item << std::endl;
next_item++;
cv_consumer.notify_one(); // 通知消费者有新数据可用
if (next_item > 40) { // 达到生产40个任务后设置退出标志
should_quit = true;
cv_consumer.notify_all(); // 唤醒所有消费者线程以检查退出条件
}
}
}
void consumer(int id) {
while (true) {
std::unique_lock<std::mutex> lock(mtx); // 获取互斥锁
cv_consumer.wait(lock, []{ return !buffer.empty() || should_quit; }); // 等待直到缓冲区不为空或需要退出
if (buffer.empty() && should_quit) break; // 如果需要退出且缓冲区为空，则退出消费者线程
int item = buffer.front(); // 从缓冲区消费一个数据
buffer.pop(); // 移除已消费的数据
std::cout << "Consumer " << id << " consumed: " << item << std::endl;
cv_producer.notify_one(); // 通知生产者缓冲区有空间
}
}
int main() {
std::thread producer1(producer, 1);
std::thread producer2(producer, 2);
std::vector<std::thread> consumers;
for (int i = 1; i <= 3; ++i) {
consumers.emplace_back(consumer, i);
}
// 等待生产者完成任务
producer1.join();
producer2.join();
// 等待消费者完成任务
for (auto& consumer : consumers) {
consumer.join();
}
return 0;
}
Producer 1 produced: 1
Consumer 3 consumed: 1
Producer 1 produced: 2
Consumer 3 consumed: 2
Producer 2 produced: 3
Consumer 2 consumed: 3
Producer 1 produced: 4
Producer 1 produced: 5
Consumer 2 consumed: 4
Consumer 2 consumed: 5
Producer 1 produced: 6
Producer 1 produced: 7
Consumer 3 consumed: 6
Consumer 3 consumed: 7
Producer 1 produced: 8
Consumer 3 consumed: 8
Producer 2 produced: 9
Consumer 2 consumed: 9
Producer 2 produced: 10
Consumer 2 consumed: 10
Producer 2 produced: 11
Consumer 3 consumed: 11
Producer 2 produced: 12
Consumer 1 consumed: 12
Producer 1 produced: 13
Consumer 2 consumed: 13
Producer 2 produced: 14
Consumer 3 consumed: 14
Producer 2 produced: 15
Consumer 2 consumed: 15
Producer 2 produced: 16
Consumer 3 consumed: 16
Producer 2 produced: 17
Producer 2 produced: 18
Consumer 2 consumed: 17
Consumer 2 consumed: 18
Producer 2 produced: 19
Consumer 3 consumed: 19
Producer 1 produced: 20
Consumer 2 consumed: 20
Producer 1 produced: 21
Consumer 2 consumed: 21
Producer 1 produced: 22
Producer 1 produced: 23
Consumer 1 consumed: 22
Consumer 1 consumed: 23
Producer 1 produced: 24
Producer 1 produced: 25
Producer 1 produced: 26
Producer 1 produced: 27
Producer 1 produced: 28
Producer 1 produced: 29
Consumer 1 consumed: 24
Consumer 1 consumed: 25
Consumer 1 consumed: 26
Consumer 1 consumed: 27
Consumer 1 consumed: 28
Consumer 1 consumed: 29
Producer 2 produced: 30
Consumer 1 consumed: 30
Producer 2 produced: 31
Consumer 1 consumed: 31
Producer 2 produced: 32
Consumer 3 consumed: 32
Producer 2 produced: 33
Producer 2 produced: 34
Consumer 1 consumed: 33
Consumer 1 consumed: 34
Producer 2 produced: 35
Consumer 3 consumed: 35
Producer 1 produced: 36
Consumer 1 consumed: 36
Producer 1 produced: 37
Consumer 1 consumed: 37
Producer 1 produced: 38
Consumer 3 consumed: 38
Producer 1 produced: 39
Consumer 2 consumed: 39
Producer 1 produced: 40
Consumer 2 consumed: 40
