生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
生产者消费者模型优点
BlockingQueue
在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)
—（图片摘于相关教材资料）
pthread_mutex_lock(&mutex);
while (条件为假）
pthread_cond_wait(cond, mutex);
修改条件
pthread_mutex_unlock(&mutex);
pthread_mutex_lock(&mutex);
设置条件为真
pthread_cond_signal(cond);
pthread_mutex_unlock(&mutex);
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#define gDefaultCap 5
template <class T>
class BlockQueue
{
bool isQueueEmpty()
{
return _bq.size() == 0;
}
bool isQueueFull()
{
return _bq.size() == _capacity;
}
public:
BlockQueue(int capacity = gDefaultCap) : _capacity(capacity)
{
pthread_mutex_init(&_mtx, nullptr);
pthread_cond_init(&_Empty, nullptr);
pthread_cond_init(&_Full, nullptr);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mtx);
pthread_cond_destroy(&_Empty);
pthread_cond_destroy(&_Full);
}
void push(const T &in) // 生产者
{
pthread_mutex_lock(&_mtx);
while (isQueueFull())
pthread_cond_wait(&_Full, &_mtx); // 检测临界资源条件
// 数据满了; 等待消费者消费数据
_bq.push(in); // 生产者放数据
//if(_bq.size() >= _capacity/2) pthread_cond_signal(&_Empty); //数据积累到>=一半时再发送
pthread_cond_signal(&_Empty); // 唤醒; 通知消费者，可以消费了
pthread_mutex_unlock(&_mtx);
}
void pop(T *out) // 消费者
{
pthread_mutex_lock(&_mtx);
while (isQueueEmpty())
pthread_cond_wait(&_Empty, &_mtx); // 检测临界资源条件
// 数据为空; 等待生产者生产数据
*out = _bq.front();
_bq.pop(); // 消费者消费
pthread_cond_signal(&_Full); // 唤醒; 通知生产者，可以生成了了
pthread_mutex_unlock(&_mtx);
}
public:
std::queue<T> _bq; // 阻塞队列
int _capacity; // 容量上限
pthread_mutex_t _mtx; // 通过互斥锁保证队列安全
pthread_cond_t _Empty; // 用它来表示bq 是否空的条件
pthread_cond_t _Full; // 用它来表示bq 是否满的条件
};
#include <iostream>
#include <pthread.h>
#include <ctime>
#include <unistd.h>
#include "BlockQueue.hpp"
void *productor(void *args)
{
BlockQueue<int> *bq = (BlockQueue<int> *)args;
while (true)
{
int a = rand() % 10 + 1;
bq->push(a);
std::cout << "productor生成的数据是" << a << std::endl;
}
return nullptr;
}
void *consumer(void *args)
{
BlockQueue<int> *bq = (BlockQueue<int> *)args;
while (true)
{
usleep(rand()%1000);
int a;
bq->pop(&a);
std::cout << "consumer消费的数据是" << a << std::endl;
sleep(1);
}
return nullptr;
}
int main()
{
// 随机数种子
srand((uint64_t)time(nullptr) ^ getpid() ^ 0x202300);
BlockQueue<int> *bq = new BlockQueue<int>();
// 创建线程
pthread_t c, p;
pthread_create(&c, nullptr, consumer, bq);
pthread_create(&p, nullptr, productor, bq);
// 等待线程
pthread_join(c, nullptr);
pthread_join(p, nullptr);
// 释放资源
delete bq;
return 0;
}
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#include "LockGuard.hpp"
#define gDefaultCap 5
template <class T>
class BlockQueue
{
bool isQueueEmpty()
{
return _bq.size() == 0;
}
bool isQueueFull()
{
return _bq.size() == _capacity;
}
public:
BlockQueue(int capacity = gDefaultCap) : _capacity(capacity)
{
pthread_mutex_init(&_mtx, nullptr);
pthread_cond_init(&_Empty, nullptr);
pthread_cond_init(&_Full, nullptr);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mtx);
pthread_cond_destroy(&_Empty);
pthread_cond_destroy(&_Full);
}
void push(const T &in) // 生产者
{
// pthread_mutex_lock(&_mtx);
LockGuard Lockguard(_mtx);
while (isQueueFull()) //访问临界资源，100%确定，资源是就绪的！
pthread_cond_wait(&_Full, &_mtx); // 检测临界资源条件
// 数据满了; 等待消费者消费数据
_bq.push(in); // 生产者放数据
//if(_bq.size() >= _capacity/2) pthread_cond_signal(&_Empty); //数据积累到>=一半时再发送
pthread_cond_signal(&_Empty); // 唤醒; 通知消费者，可以消费了
//pthread_mutex_unlock(&_mtx);
}// 出了函数后自动调用lockgrard 析构函数
void pop(T *out) // 消费者
{
// pthread_mutex_lock(&_mtx);
LockGuard Lockguard(_mtx);
while (isQueueEmpty())
pthread_cond_wait(&_Empty, &_mtx); // 检测临界资源条件
// 数据为空; 等待生产者生产数据
*out = _bq.front();
_bq.pop(); // 消费者消费
pthread_cond_signal(&_Full); // 唤醒; 通知生产者，可以生成了了
// pthread_mutex_unlock(&_mtx);
}
public:
std::queue<T> _bq; // 阻塞队列
int _capacity; // 容量上限
pthread_mutex_t _mtx; // 通过互斥锁保证队列安全
pthread_cond_t _Empty; // 用它来表示bq 是否空的条件
pthread_cond_t _Full; // 用它来表示bq 是否满的条件
};
#pragma once
#include <iostream>
#include <functional>
using func_t = std::function<int(int, int)>;
class Task
{
public:
Task() {} // 便于获取任务
Task(int x, int y, func_t func) : _x(x), _y(y), _func(func) // 制作任务
{}
int operator()() // 仿函数
{
return _func(_x, _y); // 函数调用
}
public:
int _x;
int _y;
func_t _func;
};
#pragma once
#include <iostream>
#include <pthread.h>
class Mutex
{
public:
Mutex(pthread_mutex_t& mtx):_mtx(mtx)
{}
void lock()
{
std::cout << "要进行加锁" << std::endl;
pthread_mutex_lock(&_mtx);
}
void unlock()
{
std::cout << "要进行解锁" << std::endl;
pthread_mutex_unlock(&_mtx);
}
~Mutex()
{}
public:
pthread_mutex_t _mtx;
};
// RAII风格的加锁方式
class LockGuard
{
public:
LockGuard(pthread_mutex_t& mtx):_mx(mtx)
{
_mx.lock();
}
~LockGuard()
{
_mx.unlock();
}
private:
Mutex _mx;
};
#include <iostream>
#include <pthread.h>
#include <ctime>
#include <unistd.h>
#include "BlockQueue.hpp"
#include "Task.hpp"
int myMul(int x, int y)
{
return x * y;
}
void *productor(void *args)
{
// BlockQueue<int> *bq = (BlockQueue<int> *)args;
BlockQueue<Task> *bq = (BlockQueue<Task> *)args;
while (true)
{
// int a = rand() % 10 + 1;
// bq->push(a);
// std::cout << "productor生成的数据是" << a << std::endl;
// 制作任务
int x = rand() % 10 + 1;
usleep(rand() % 1000);
int y = rand() % 5 + 1;
std::cout << pthread_self() << ":发布任务..." << x << 'x' << y << "=" << '?' << std::endl;
Task t(x, y, myMul);
bq->push(t);
sleep(1);
}
return nullptr;
}
void *consumer(void *args)
{
// BlockQueue<int> *bq = (BlockQueue<int> *)args;
BlockQueue<Task> *bq = (BlockQueue<Task> *)args;
while (true)
{
// usleep(rand() % 1000);
// int a;
// bq->pop(&a);
// std::cout << "consumer消费的数据是" << a << std::endl;
// sleep(1);
// 获取任务
Task t;
bq->pop(&t);
// 完成任务
std::cout << pthread_self() << ":执行任务..." << t._x << 'x' << t._y << "=" << t() << std::endl;
}
return nullptr;
}
#define CONSUMER_NUM 6
#define PRODUCTOR_NUM 3
int main()
{
// 随机数种子
srand((uint64_t)time(nullptr) ^ getpid() ^ 0x202300);
// BlockQueue<int> *bq = new BlockQueue<int>();
BlockQueue<Task> *bq = new BlockQueue<Task>();
// 创建线程
pthread_t c[CONSUMER_NUM], p[PRODUCTOR_NUM];
for(int i=0; i<CONSUMER_NUM; ++i)
{
pthread_create(c+i, nullptr, consumer, bq);
}
for(int i=0; i<PRODUCTOR_NUM; ++i)
{
pthread_create(p+i, nullptr, productor, bq);
}
// 等待线程
for(int i=0; i<CONSUMER_NUM; ++i)
{
pthread_join(c[i], nullptr);
}
for(int i=0; i<PRODUCTOR_NUM; ++i)
{
pthread_join(p[i], nullptr);
}
// 释放资源
delete bq;
return 0;
}
