【Linux学习笔记】线程同步与互斥之生产者消费者模型

【Linux学习笔记】线程同步与互斥之生产者消费者模型

🔥个人主页：大白的编程日记

🔥专栏：Linux学习笔记

---

前言

哈喽，各位小伙伴大家好!上期我们讲了线程同步与互斥(一) 今天我们讲的是线程同步与互斥之生产者消费者模型。话不多说，我们进入正题！向大厂冲锋！

2. 线程同步

2-1条件变量

• 当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
• 例如一个线程访问队列时，发现队列为空，它只能等待，只到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。
  

2-2 同步概念与竞态条件

• 同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步
• 竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理解

2-3条件变量函数

初始化

1 intPTHread_cond_init(pthread_cond_t \*restrict cond,constPTHread_condattr_t \*restrict attr);  
2 参数：  
3 cond: 要初始化的条件变量  
4 attr: NULL

销毁

1 intPTHread_cond destroy(pthread_cond_t \*cond)

等待条件满足

1 int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);  
2 参数：  
3 cond: 要在这个条件变量上等待  
4 mutex: 互斥量, 后面详细解释

1 int pthread_cond_broadcast(pthread_cond_t *cond);
2 int pthread_cond_signal(pthread_cond_t *cond);

简单案例：

• 我们先使用PTHREAD_COND/MUTEX_INITIALIZER进行测试，对其他细节暂不追究
• 然后将接口更改成为使用 pthread_cond_init/pthread_cond_destroy 的方式，方便后续进行封装

#include <iostream>   
#include <string.h>   
#include <unistd.h>   
#include <pthread.h>   
pthread_cond_t cond = PTHREAD_COND Initialized;   
pthread_mutex_t mutex = PTHREAD_MUTEX Initialized;   
void *active(void *arg)   
{   std::string name = static_cast(const char*>(arg);   while (true) {   pthread_mutex_lock(&_mutex);   pthread_cond_wait(&cond, &_mutex);   std::cout << name << "活动..." << std::endl;   pthread_mutex_unlock(&_mutex);   
}   
}   
int main(void)   
{   pthread_t t1, t2;   pthread_create(&t1, NULL, active, (void*)"thread-1");   pthread_create(&t2, NULL, active, (void*)"thread-2");   sleep(3); //可有可无，这里确保两个线程已经在运行  while(true)   {   //对比测试   //pthread_cond_signal(&cond); //唤醒一个线程   pthread_cond_broadcast(&cond); //唤醒所有线程   sleep(1);   
}pthread_join(t1, NULL);pthread_join(t2, NULL);}

1 $ ./cond
2 thread-1 活动...
3 thread-2 活动...
4 thread-1 活动...
5 thread-1 活动...
6 thread-2 活动...

2-4生产者消费者模型

321原则(便于记忆)

2-4-1 为何要使用生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

2-4-2 生产者消费者模型优点

• 解耦
• 支持并发
• 支持忙闲不均

2-5基于BlockingQueue的生产者消费者模型

2-5-1 BlockingQueue

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)

2-5-2 C++ queue模拟阻塞队列的生产消费模型

代码：

• 我们以单生产者，单消费者，来进行讲解。
  • 刚开始写，我们采用原始接口。
• 我们先写单生产，单消费。然后改成多生产，多消费（这里代码其实不变）。

BlockQueue.hpp

#pragma once
#include <pthread.h>
#include <iostream>
#include <string>
#include <queue>
using namespace std;const int defaultcap = 5; // for test
template <class T>
class BlockQueue
{
public:bool IsFull(){return _cap <= _q.size();}bool IsEmpty(){return _q.size() == 0;}BlockQueue(int cap = defaultcap): _cap(cap), _csleep_num(0), _psleep_num(0){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_full_cond, nullptr);pthread_cond_init(&_empty_cond, nullptr);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_full_cond);pthread_cond_destroy(&_empty_cond);}T Pop(){pthread_mutex_lock(&_mutex);// 避免休眠失败向后直接pop 进行二次检查！while (IsEmpty()){// 队列为空消费者去条件变量下休眠_csleep_num++;pthread_cond_wait(&_empty_cond, &_mutex);_csleep_num--;}T ret = _q.front();_q.pop();// 走到这一定有生产空间 唤醒生产者生产if (_psleep_num){pthread_cond_signal(&_full_cond);cout << "唤醒一个生产者" << endl;}pthread_mutex_unlock(&_mutex);return ret;}void Equeue(const T& date){pthread_mutex_lock(&_mutex);// 避免休眠失败向后直接push 进行二次检查！while (IsFull()){cout<<"生产者进行休眠了！:"<<_psleep_num<<endl;// 队列为满生产者去条件变量下休眠_psleep_num++;pthread_cond_wait(&_full_cond, &_mutex);_psleep_num--;}_q.push(date);// 走到这一定有消费空间 唤醒消费者生产if (_csleep_num){pthread_cond_signal(&_empty_cond);cout << "唤醒一个消费者" << endl;}pthread_mutex_unlock(&_mutex);pthread_cond_signal(&_empty_cond);}private:std::queue<T> _q; // 临界资源！！！int _cap;         // 容量大小pthread_mutex_t _mutex;pthread_cond_t _full_cond;pthread_cond_t _empty_cond;int _csleep_num; // 消费者休眠的个数int _psleep_num; // 生产者休眠的个数
};

注意：这里采用模版，是想告诉我们，队列中不仅仅可以防止内置类型，比如int,对象也可以作为任务来参与生产消费的过程哦。

1 #pragma once   
2   
3 #include<iostream>   
4 #include<string>   
5 #include<functional>   
6   
7 //任务类型1   
8 //class Task   
9 //{   
10 //public:   
11 //Task(){}   
12 //Task(int a,int b):_a(a),_b(b),_result(0)
13 // {  
14 // }  
15 // void Excute()  
16 // {  
17 // _result = _a + _b;  
18 // }  
19 // std::string ResultToString()  
20 // {  
21 // return std::to_string(_a) + "+" + std::to_string(_b) + "=" + std::to_string(_result);  
22 // }  
23 // std::string DebugToString()  
24 // {  
25 // return std::to_string(_a) + "+" + std::to_string(_b) + "=";  
26 // }  
27  
28 // private:  
29 // int _a;  
30 // int _b;  
31 // int _result;  
32 // };  
33  
34 // 任务类型2  
35 using Task = std::function<void>();

2-6 为什么 pthread_cond_wait 需要互斥量？

• 条件等待是线程间同步的一种手段，如果只有一个线程，条件不满足，一直等下去都不会满足，所以必须要有一个线程通过某些操作，改变共享变量，使原先不满足的条件变得满足，并且友好的通知等待在条件变量上的线程。
• 条件不会无缘无故的突然变得满足了，必然会牵扯到共享数据的变化。所以一定要用互斥锁来保护。没有互斥锁就无法安全的获取和修改共享数据。

互斥量

被锁挡住了没办法执行

造成死锁

• 按照上面的说法，我们设计出如下的代码：先上锁，发现条件不满足，解锁，然后等待在条件变量上不就行了，如下代码：

// 错误的设计
pthread_mutex_lock(&mutex);
while (condition_is_false) {pthread_mutex_unlock(&mutex);// 解锁之后，等待之前，条件可能已经满足，信号已经发出，但是该信号可能被错过pthread_cond_wait(&cond, &mutex);pthread_mutex_lock(&mutex);
}
pthread_mutex_unlock(&mutex);

• 由于解锁和等待不是原子操作。调用解锁之后，pthread_cond_wait 之前，如果已经有其他线程获取到互斥量，摒弃条件满足，发送了信号，那么pthread_cond_wait 将错过这个信号，可能会导致线程永远阻塞在这个pthread_cond_wait。所以解锁和等待必须是一个原子操作。
• intPTHread_cond_wait(pthread_cond_t \*cond,pthread_mutex_t\* mutex)；进入该函数后，会去看条件量等于0不？等于，就把互斥量变成1，直到cond_wait返回，把条件量改成1，把互斥量恢复成原样。

2-7条件变量使用规范

等待条件代码

1 pthread_mutex_lock(& mutex);   
2 while（条件为假）//if??   
3 pthread_cond_waitcond，mutex);   
4 修改条件   
5 pthread_mutex_unlock(&mutex);

• 给条件发送信号代码

1 pthread_mutex_lock(& mutex);   
2 设置条件为真   
3 pthread_cond_signalcond);   
4 pthread_mutex_unlock(& mutex);

2-8条件变量的封装

• 基于上面的基本认识，我们已经知道条件变量如何使用，虽然细节需要后面再来进行解释，但这里可以做一下基本的封装，以备后用。

Cond.hpp

#include <pthread.h>
#include <iostream>
using namespace std;
#include"Mutex.hpp"
using namespace MutexModule;
namespace CondModule
{class Cond{public:Cond(){pthread_cond_init(&_cond,nullptr);}~Cond(){pthread_cond_destroy(&_cond);}void Wait(Mutex& mutex){pthread_cond_wait(&_cond,mutex.Get());}void Signal(){pthread_cond_signal(&_cond);}void Brodcast(){pthread_cond_broadcast(&_cond);}private:pthread_cond_t _cond;};
}

后言

这就是线程同步与互斥之生产者消费者模型。大家自己好好消化！今天就分享到这！ 感谢各位的耐心垂阅！咱们下期见！拜拜~