目录
(1)生产者消费者模型是多线程同步与互斥的一个经典场景，其特点如下：
我们用代码编写生产者消费者模型的时候，本质就是对这三个特点进行维护。
(2)生产者和生产者、消费者和消费者、生产者和消费者，它们之间为什么会存在互斥关系？
(3)生产者和消费者之间为什么会存在同步关系？
注意： 互斥关系保证的是数据的正确性，而同步关系是为了让多线程之间协同起来。
(4)让消费者和生产者协同工作，合适的时候可能一直运行，生产者和消费者并不会因为要互相等待对方的结果而阻塞，相当于双方可以并发执行.
(5)为什么要有生产者消费者模型？ 本质是用代码进行解耦的过程
(1)如果我们在主函数中调用某一函数，那么我们必须等该函数体执行完后才继续执行主函数的后续代码，因此函数调用本质上是一种紧耦合。
(2)对应到生产者消费者模型中，函数传参实际上就是生产者生产的过程，而执行函数体实际上就是消费者消费的过程，但生产者只负责生产数据，消费者只负责消费数据，在消费者消费期间生产者可以同时进行生产，因此生产者消费者模型本质是一种松耦合。
(1) 生产和消费步调一致，生产一个消费一个
①BlockQueue.hpp
#pragma oncec
#include <iostream>
#include <pthread.h>
#include <queue>
#include <unistd.h>
#define NUM 5
template<class T>
class BlockQueue
{
private:
bool IsFull()
{
return _q.size() == _cap;
}
bool IsEmpty()
{
return _q.empty();
}
public:
BlockQueue(int cap = NUM)
: _cap(cap)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_full, nullptr);
pthread_cond_init(&_empty, nullptr);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_full);
pthread_cond_destroy(&_empty);
}
public:
//向阻塞队列插入数据（生产者调用）
void Push(const T& data)
{
pthread_mutex_lock(&_mutex);
while (IsFull()){
//不能进行生产，直到阻塞队列可以容纳新的数据
pthread_cond_wait(&_full, &_mutex);
}
_q.push(data);
pthread_cond_signal(&_empty); //唤醒在empty条件变量下等待的消费者线程
pthread_mutex_unlock(&_mutex);
}
//从阻塞队列获取数据（消费者调用）
void Pop(T& data)
{
pthread_mutex_lock(&_mutex);
while (IsEmpty()){
//不能进行消费，直到阻塞队列有新的数据
pthread_cond_wait(&_empty, &_mutex);
}
data = _q.front();
_q.pop();
pthread_cond_signal(&_full); //唤醒在full条件变量下等待的生产者线程
pthread_mutex_unlock(&_mutex);
}
private:
std::queue<T> _q; //阻塞队列
int _cap; //阻塞队列容纳数据的最大个数
pthread_mutex_t _mutex;
pthread_cond_t _full;
pthread_cond_t _empty;
};
程序说明:
判断是否满足生产消费条件时不能用if，而应该用while：
②main.cc
#include "BlockQueue.hpp"
void* Producer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//生产者不断进行生产
while (true){
sleep(1);
int data = rand() % 100 + 1;
bq->Push(data); //生产数据
std::cout << "Producer: " << data << std::endl;
}
}
void* Consumer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//消费者不断进行消费
while (true){
sleep(1);
int data = 0;
bq->Pop(data); //消费数据
std::cout << "Consumer: " << data << std::endl;
}
}
int main()
{
srand((unsigned int)time(nullptr));//生成随机数
pthread_t producer, consumer;
BlockQueue<int>* bq = new BlockQueue<int>;
//创建生产者线程和消费者线程
pthread_create(&producer, nullptr, Producer, bq);
pthread_create(&consumer, nullptr, Consumer, bq);
//join生产者线程和消费者线程
pthread_join(producer, nullptr);
pthread_join(consumer, nullptr);
delete bq;
return 0;
}程序说明:
③结果 : 生产者是每隔一秒生产一个数据，所以消费者是每隔一秒消费一个数据，因此运行代码后我们可以看到生产者和消费者的执行步调是一致的。
(2)生产快，消费慢
①代码: 只需要改变main.cc中的执行函数
void* Producer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//生产者不断进行生产
while (true){
int data = rand() % 100 + 1;
bq->Push(data); //生产数据
std::cout << "Producer: " << data << std::endl;
}
}
void* Consumer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//消费者不断进行消费
while (true){
sleep(1);
int data = 0;
bq->Pop(data); //消费数据
std::cout << "Consumer: " << data << std::endl;
}
}②结果: 生产者生产的很快，运行后一瞬间生产者就将阻塞队列打满了，此时生产者想要再进行生产就只能在full条件变量下进行等待，直到消费者消费完一个数据后，生产者才会被唤醒进而继续进行生产，生产者生产完一个数据后又会进行等待，因此后续生产者和消费者的步调又变成一致的了 ; 顺序消费。
(3)生产慢，消费快
①代码
void* Producer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//生产者不断进行生产
while (true){
sleep(1);
int data = rand() % 100 + 1;
bq->Push(data); //生产数据
std::cout << "Producer: " << data << std::endl;
}
}
void* Consumer(void* arg)
{
BlockQueue<int>* bq = (BlockQueue<int>*)arg;
//消费者不断进行消费
while (true){
int data = 0;
bq->Pop(data); //消费数据
std::cout << "Consumer: " << data << std::endl;
}
}②结果 : 虽然消费者消费的很快，但一开始阻塞队列中是没有数据的，因此消费者只能在empty条件变量下进行等待，直到生产者生产完一个数据后，消费者才会被唤醒进而进行消费，消费者消费完这一个数据后又会进行等待，因此生产者和消费者的步调就是一致的。
(4) 生产/消费超过一定高低水位线再通知另一方
①代码 : 生产者生产快不sleep，消费者消费慢sleep
//向阻塞队列插入数据（生产者调用）
void Push(const T& data)
{
pthread_mutex_lock(&_mutex);
while (IsFull()){
//不能进行生产，直到阻塞队列可以容纳新的数据
pthread_cond_wait(&_full, &_mutex);
}
_q.push(data);
//高低水位线
if (_q.size() >= _cap / 2){
pthread_cond_signal(&_empty); //唤醒在empty条件变量下等待的消费者线程
}
pthread_mutex_unlock(&_mutex);
}
//从阻塞队列获取数据（消费者调用）
void Pop(T& data)
{
pthread_mutex_lock(&_mutex);
while (IsEmpty()){
//不能进行消费，直到阻塞队列有新的数据
pthread_cond_wait(&_empty, &_mutex);
}
data = _q.front();
_q.pop();
//高低水位线
if (_q.size() <= _cap / 2){
pthread_cond_signal(&_full); //唤醒在full条件变量下等待的生产者线程
}
pthread_mutex_unlock(&_mutex);
}
②结果 :运行代码后生产者还是一瞬间将阻塞队列打满后进行等待，但此时不是消费者消费一个数据就唤醒生产者线程，而是当阻塞队列当中的数据小于队列容器的一半时，才会唤醒生产者线程进行生产。
(1)代码
①Task.hpp : 定义一个Task类，这个类当中包含一个Run成员函数，该函数代表着如何让消费者处理拿到的数据。
#pragma once
#include <iostream>
class Task
{
public:
Task(int x = 0, int y = 0, int op = 0)
: _x(x), _y(y), _op(op)
{}
~Task()
{}
void Run()
{
int result = 0;
switch (_op)
{
case '+':
result = _x + _y;
break;
case '-':
result = _x - _y;
break;
case '*':
result = _x * _y;
break;
case '/':
if (_y == 0){
std::cout << "Warning: div zero!" << std::endl;
result = -1;
}
else{
result = _x / _y;
}
break;
case '%':
if (_y == 0){
std::cout << "Warning: mod zero!" << std::endl;
result = -1;
}
else{
result = _x % _y;
}
break;
default:
std::cout << "error operation!" << std::endl;
break;
}
std::cout << _x << _op << _y << "=" << result << std::endl;
}
private:
int _x;
int _y;
char _op;
};
②main.cc : 对线程执行函数稍作修改, 此时生产者放入阻塞队列的数据就是一个Task对象，而消费者从阻塞队列拿到Task对象后，就可以用该对象调用Run成员函数进行数据处理。
void* Producer(void* arg)
{
BlockQueue<Task>* bq = (BlockQueue<Task>*)arg;
const char* arr = "+-*/%";
//生产者不断进行生产
while (true){
int x = rand() % 100;
int y = rand() % 100;
char op = arr[rand() % 5];
Task t(x, y, op);
bq->Push(t); //生产数据
std::cout << "producer task done" << std::endl;
}
}
void* Consumer(void* arg)
{
BlockQueue<Task>* bq = (BlockQueue<Task>*)arg;
//消费者不断进行消费
while (true){
sleep(1);
Task t;
bq->Pop(t); //消费数据
t.Run(); //处理数据
}
}
(2)结果 : 运行代码，当阻塞队列被生产者打满后消费者被唤醒，此时消费者在消费数据时执行的就是计算任务，当阻塞队列当中的数据被消费到低水位线时又会唤醒生产者进行生产。
(3)小结
我们想让生产者消费者模型处理某一种任务时，大体的框架已经搭建好了，就只需要提供对应的Task类，然后让该Task类提供一个对应的Run成员函数告诉我们应该如何处理这个任务即可
(1)为什么要有信号量？ 提高效率
(2)信号量的PV操作：
(3)PV操作必须是原子操作
(4)申请信号量失败被挂起等待
(5)信号量结构的大致理解伪代码:
(1) 初始化
函数 : int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
- sem：需要初始化的信号量。
- pshared：传入0值表示线程间共享，传入非零值表示进程间共享(常用0)
- value：信号量的初始值（计数器的初始值）。
返回值： 初始化信号量成功返回0，失败返回-1。
注意： POSIX信号量和System V信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的，但POSIX信号量可以用于线程间同步。
(2)销毁
函数 : int sem_destroy(sem_t *sem);
参数： sem：需要销毁的信号量。
返回值：销毁信号量成功返回0，失败返回-1。
(3)等待信号量（申请信号量 ,P()）
函数 : int sem_wait(sem_t *sem);
参数：sem：需要等待的信号量。
返回值：
- 等待信号量成功返回0，信号量的值减一
- 等待信号量失败返回-1，信号量的值保持不变
(4)发布信号量（释放信号量,V()）
函数 : int sem_post(sem_t *sem);
参数：sem：需要发布的信号量。
返回值：
- 发布信号量成功返回0，信号量的值加一
- 发布信号量失败返回-1，信号量的值保持不变
示例，实现一个多线程抢票系统，使用二元信号量模拟实现多线程互斥,，让每个线程在访问全局变量tickets之前先申请信号量，访问完毕后再释放信号量，此时二元信号量就达到了互斥的效果。
1.代码
#include <iostream>
#include <string>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
class Sem{
public:
Sem(int num)
{
sem_init(&_sem, 0, num);
}
~Sem()
{
sem_destroy(&_sem);
}
void P()
{
sem_wait(&_sem);
}
void V()
{
sem_post(&_sem);
}
private:
sem_t _sem;
};
Sem sem(1); //二元信号量
int tickets = 1000;
void* TicketGrabbing(void* arg)
{
std::string name = (char*)arg;
while (true){
sem.P();
if (tickets > 0){
usleep(1000);
std::cout << name << " get a ticket, tickets left: " << --tickets << std::endl;
sem.V();
}
else{
sem.V();
break;
}
}
std::cout << name << " quit..." << std::endl;
pthread_exit((void*)0);
}
int main()
{
pthread_t tid1, tid2, tid3, tid4;
pthread_create(&tid1, nullptr, TicketGrabbing, (void*)"thread 1");
pthread_create(&tid2, nullptr, TicketGrabbing, (void*)"thread 2");
pthread_create(&tid3, nullptr, TicketGrabbing, (void*)"thread 3");
pthread_create(&tid4, nullptr, TicketGrabbing, (void*)"thread 4");
pthread_join(tid1, nullptr);
pthread_join(tid2, nullptr);
pthread_join(tid3, nullptr);
pthread_join(tid4, nullptr);
return 0;
}2.结果 : 未出现数据不一致问题
(1)生产者关注的是空间资源，消费者关注的是数据资源
(2)blank_sem和data_sem的初始值设置
现在我们用信号量来描述环形队列当中的空间资源（blank_sem）和数据资源（data_sem），在我们初始信号量时给它们设置的初始值是不同的：
(3)生产者和消费者申请和释放资源
①生产者申请空间资源，释放数据资源
对于生产者来说，生产者每次生产数据前都需要先申请blank_sem：
当生产者生产完数据后，应该释放data_sem：
②消费者申请数据资源，释放空间资源
对于消费者来说，消费者每次消费数据前都需要先申请data_sem：
当消费者消费完数据后，应该释放blank_sem：
(1)生产者和消费者不能对同一个位置进行访问(互斥)
(2)无论是生产者还是消费者，都不应该将对方套一个圈以上(格子的数量有限)
(1) 代码
①RingQueue.hpp
#pragma once
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <vector>
#define NUM 10
template<class T>
class RingQueue
{
private:
//P操作
void P(sem_t& s)
{
sem_wait(&s);
}
//V操作
void V(sem_t& s)
{
sem_post(&s);
}
public:
RingQueue(int cap = NUM)
: _cap(cap), _p_pos(0), _c_pos(0)
{
_q.resize(_cap);
sem_init(&_blank_sem, 0, _cap); //blank_sem初始值设置为环形队列的容量
sem_init(&_data_sem, 0, 0); //data_sem初始值设置为0
}
~RingQueue()
{
sem_destroy(&_blank_sem);
sem_destroy(&_data_sem);
}
public:
//向环形队列插入数据（生产者调用）
void Push(const T& data)
{
P(_blank_sem); //生产者关注空间资源
_q[_p_pos] = data;
V(_data_sem); //生产
//更新下一次生产的位置
_p_pos++;
_p_pos %= _cap;
}
//从环形队列获取数据（消费者调用）
void Pop(T& data)
{
P(_data_sem); //消费者关注数据资源
data = _q[_c_pos];
V(_blank_sem);
//更新下一次消费的位置
_c_pos++;
_c_pos %= _cap;
}
private:
std::vector<T> _q; //环形队列
int _cap; //环形队列的容量上限
int _p_pos; //生产位置
int _c_pos; //消费位置
sem_t _blank_sem; //描述空间资源
sem_t _data_sem; //描述数据资源
};程序解释:
②main.cc实现单生产者，单消费者模型
#include "RingQueue.hpp"
void* Producer(void* arg)
{
RingQueue<int>* rq = (RingQueue<int>*)arg;
while (true){
sleep(1);
int data = rand() % 100 + 1;
rq->Push(data);
std::cout << "Producer: " << data << std::endl;
}
}
void* Consumer(void* arg)
{
RingQueue<int>* rq = (RingQueue<int>*)arg;
while (true){
sleep(1);
int data = 0;
rq->Pop(data);
std::cout << "Consumer: " << data << std::endl;
}
}
int main()
{
srand((unsigned int)time(nullptr));
pthread_t producer, consumer;
RingQueue<int>* rq = new RingQueue<int>;
pthread_create(&producer, nullptr, Producer, rq);
pthread_create(&consumer, nullptr, Consumer, rq);
pthread_join(producer, nullptr);
pthread_join(consumer, nullptr);
delete rq;
return 0;
}
(2)结果
①生产者消费者同步
②生产快消费慢
③生产慢消费快
(1)在blank_sem和data_sem两个信号量的保护后，该环形队列中不可能会出现数据不一致的问题。
①因为只有当生产者和消费者指向同一个位置并访问时，才会导致数据不一致的问题，而此时生产者和消费者在对环形队列进行写入或读取数据时，只有两种情况会指向同一个位置：
②但是在这两种情况下，生产者和消费者不会同时对环形队列进行访问：
(2)当环形队列为空和满时，我们已经通过信号量保证了生产者和消费者的串行化过程。而除了这两种情况之外，生产者和消费者指向的都不是同一个位置，因此该环形队列当中不可能会出现数据不一致的问题。并且大部分情况下生产者和消费者指向并不是同一个位置，因此大部分情况下该环形队列可以让生产者和消费者并发的执行.