目录
概念及优势
代码实现
概念及优势
生产者消费者模型是一种用于线程同步的模型，在这个模型中有两种角色，生产者生产数据，消费者消费数据。有三种关系，生产者与生产者，消费者与消费者，生产者与消费者。还有一个交易场所。
超市就是生活中最常见的生产者消费者模型，工厂生产商品，超市充当缓冲区，消费者去超市消费同时取走超市中的商品。
超市作为缓冲区，起到了很重要的作用，试想如果没有超市，那消费者想购物只能去找工厂，还要等待工厂将商品生产出来，同时工厂也不能持续生产商品，因为没有足够的空间存放，而超市作为缓冲区解决了这些问题，消费者不用去工厂阻塞等待商品生产，工厂也可以一直生产，只要超市还能放得下。
从上面的例子可以看出生产者消费者模型的优点，将生产者和消费者解耦，支持并发，支持忙闲不均。
再来讨论一下这三者间的关系
消费者与消费者不能同时向交易场所写数据，生产者与生产者也不能同时向交易场所拿数据，这两者都是互斥且竞争的关系。而生产者不能与消费者同时拿或写数据，这两者属于互斥且同步的关系。
交易场所的问题，交易场所为空时消费者不能消费，交易场所满时生产者不能生产。
代码实现
基于阻塞队列来实现生产者消费者模型，阻塞队列与普通队列的区别在于，阻塞队列在队列为空是pop操作会阻塞，队列为满时，push操作会阻塞。
//block_queue.hpp
#include <queue>
#include <pthread.h>
namespace my
{
template<class T>
class block_queue
{
public:
block_queue(size_t max_size = 1000)
:_max_size(max_size)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_consumer_cond, nullptr);
pthread_cond_init(&_producer_cond, nullptr);
}
bool empty() const
{
return _b_queue.empty();
}
bool full() const
{
return _b_queue.size() == _max_size;
}
//生产者
void push(const T& data)
{
pthread_mutex_lock(&_mutex); //加锁
//循环判断以防止虚假唤醒
//如果用if判断，试想这种场景
//线程a在等待，然后被唤醒了，既然被唤醒说明此时应该是不为满
//但是被唤醒后还需要竞争锁（因为等待会释放锁），只有线程a再次持有锁
//才会醒来执行后面的代码
//但是线程a有可能并没有立马拿到锁，被另一个生产者线程b拿到锁
//并且b此时判断也是不为满
//那么b会向后执行，会向队列里push数据
//这就有问题了，在这之后线程a拿到了锁，向后执行，但现在已经是满队列了
while(full())
{
//队列满了，生产者等待
//等待必须解锁，否则其他线程都拿不到锁
//条件变量等待的条件一定属于临界资源，
//所以pthread_cond_wait一定属于临界区代码
//所以pthread_cond_wait这个函数在内部一定会解锁来防止死锁
//被唤醒后会竞争锁
pthread_cond_wait(&_producer_cond, &_mutex);
}
_b_queue.push(data); //入队
std::cout << "生产数据" << data << std::endl;
pthread_mutex_unlock(&_mutex); //解锁
pthread_cond_signal(&_consumer_cond); //唤醒消费者
std::cout << "唤醒消费者" << std::endl;
}
const T& pop()
{
pthread_mutex_lock(&_mutex); //加锁
//循环判断以防止虚假唤醒
while(empty())
{
pthread_cond_wait(&_consumer_cond, &_mutex);
}
T& ret = _b_queue.front();
std::cout << "消费数据" << ret << std::endl;
_b_queue.pop(); //出队
pthread_mutex_unlock(&_mutex); //解锁
pthread_cond_signal(&_producer_cond); //唤醒生产者
std::cout << "唤醒生产者" << std::endl;
return ret;
}
~block_queue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_consumer_cond);
pthread_cond_destroy(&_producer_cond);
}
size_t size() const
{
return _b_queue.size();
}
private:
std::queue<T> _b_queue; //底层容器
size_t _max_size; //队列最大容量
pthread_mutex_t _mutex; //给队列加锁
pthread_cond_t _consumer_cond; //消费者条件变量
pthread_cond_t _producer_cond; //生产者条件变量
};
}main函数
#include <iostream>
#include <memory>
#include <unistd.h>
#include "block_queue.hpp"
using data = int;
void* producer(void* _bq)
{
my::block_queue<data>& bq = *(my::block_queue<data>*)_bq;
data val = 0;
while (true)
{
//使得生产者总是慢于消费者，这样队列一直都只会有一个元素
//sleep(2);
bq.push(val);
val++;
}
}
void* consumer(void* _bq)
{
my::block_queue<data>& bq = *(my::block_queue<data>*)_bq;
data val = 0;
while (true)
{
val = bq.pop();
}
}
void test()
{
pthread_t tid_consumer1 = 0, tid_producer1 = 0;
pthread_t tid_consumer2 = 1, tid_producer2 = 0;
pthread_t tid_consumer3 = 2, tid_producer3 = 0;
std::shared_ptr<my::block_queue<int>> bq(std::make_shared<my::block_queue<int>>(5));
pthread_create(&tid_producer1, nullptr, producer, &(*bq));
pthread_create(&tid_producer2, nullptr, producer, &(*bq));
pthread_create(&tid_producer3, nullptr, producer, &(*bq));
pthread_create(&tid_consumer1, nullptr, consumer, &(*bq));
pthread_create(&tid_consumer2, nullptr, consumer, &(*bq));
pthread_create(&tid_consumer3, nullptr, consumer, &(*bq));
pthread_join(tid_producer1, nullptr);
pthread_join(tid_producer2, nullptr);
pthread_join(tid_producer3, nullptr);
pthread_join(tid_consumer1, nullptr);
pthread_join(tid_consumer2, nullptr);
pthread_join(tid_consumer3, nullptr);
}
int main()
{
test();
return 0;
}
