前面我们已经提过线程互斥的相关概念，但是我们在前文的抢票逻辑中，我们其实很容易发现一个问题。那就是票可能被一直被一个人抢，这里我们就需要引入条件变量的概念。
目录
1、条件变量
<1>线程同步
<2>相关概念
<3>相关的接口
2、生产消费模型
<1>概念的引入
<2>基于条件变量的生产消费模型的示例代码
<3>多生产多消费模型
<4>小结
3、基于环形队列的生产消费模型
<1>概念引入
<2>具体介绍
<3>具体实现方法 - 信号量
<4>基于信号量的环形队列生产消费模型的代码示例
1、条件变量
<1>线程同步
我们先用一个例子引入条件变量。假设我们现在在学校里有一台冰柜，冰柜上有一把锁，每次冰柜只能被一个人使用。现在有一名同学放了一大杯饮料在冰柜中，其他同学看到冰柜被使用，就只能等第一名同学用完后，把锁交出来。但是这里就可能会出现这种问题，每当第一名用锁把冰柜打开后，把里面的冷饮喝了一口又放回去锁上了，这就导致其他同学压根拿不到锁，无法使用冰柜。学校的管理者，看到这种现象肯定是不允许的，所以就规定了每次使用完冰柜后，需要将钥匙交出，并且到队列尾部重新排队等待锁。这样就能使得每个人都有机会获得冰柜的使用权。
上述例子中的同学我们其实就可以看成是线程，而冰柜就是一个临界资源。上述例子中，修改后冰柜的使用方法，基本做到让不同线程访问同一个临界资源安全的情况下，让不同的线程访问临界资源具有顺序性，我们称这为线程同步。
<2>相关概念
如何实现线程的同步，这里我们就需要是用条件变量了。这里再用一个例子来理解一下条件变量是什么东西。
假设先在D向一个箱子内放置苹果，箱子有锁，A、B、C都会在箱子内取东西。现在箱子内没有苹果，需要D去放置苹果。但是D的抢锁能力很差，根本就拿不到锁，而A、B、C抢锁能力很强。这就会导致C一直抢锁，检查箱子内有没有苹果(箱子内没有苹果)，而D无法持有锁去放苹果。为了避免这种情况，我们就可以让A、B、C先拿锁，如果打开箱子后没有苹果(有就直接拿)，就到一旁的队列排队等待，D放完苹果后，D(用铃铛)通知了再去拿锁开箱，取出苹果。(如下图)
这里的"铃铛+队列"其实就是条件变量。
<3>相关的接口
这里三个接口的使用和互斥锁的使用一样，只不过需要定义的类型变成了pthread_cond_t而已。
pthread_cond_wait其实就和上图中队列对应，表示在该条件变量下等待。pthread_cond_timedwait和pthread_cond_wait的区别就是pthread_cond_timedwait在第三个参数处设置了过期时间，超过该时间后，线程会自动被唤醒，这个接口我们一般不使用。
这两个接口就类似于上面的铃铛，用于唤醒队列中的线程。pthread_cond_signal 表示唤醒一个线程，pthread_cond_broadcast是唤醒cond下等待的所有线程。
上述的所有接口成功都返回零，失败返回错误码
下面简单地示范一下如何使用条件变量
#include <iostream>
#include <vector>
#include <pthread.h>
#include <unistd.h>
#include <string>#define Num 4pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t g_cond = PTHREAD_COND_INITIALIZER;
bool stop_threads = false;void *Slaverwork(void *arg)
{while (true){pthread_mutex_lock(&g_mutex);while (!stop_threads){pthread_cond_wait(&g_cond, &g_mutex);}pthread_mutex_unlock(&g_mutex);break; // Exit the thread}std::string threadname = static_cast<const char *>(arg);std::cout << "线程被唤醒, 唤醒名称为： " << threadname << std::endl;delete[] (char *)arg;return nullptr;
}void *Create_Slaver(void *arg)
{std::vector<pthread_t> *tids = static_cast<std::vector<pthread_t> *>(arg);for (int i = 0; i < Num; i++){pthread_t tid;char *ptr = new char[64];snprintf(ptr, 64, "pthread - %d", i + 1);pthread_create(&tid, nullptr, Slaverwork, ptr);tids->push_back(tid);}return nullptr;
}void MasterStart(std::vector<pthread_t> &tids)
{pthread_t tid;int n = pthread_create(&tid, nullptr, Create_Slaver, &tids);if (n == 0){std::cout << "Master create success" << std::endl;}tids.push_back(tid);
}void Master_awake(std::vector<pthread_t> &tids)
{int count = 6;std::cout << "Master begin work" << std::endl;while (count--){sleep(1);pthread_mutex_lock(&g_mutex);pthread_cond_signal(&g_cond);pthread_mutex_unlock(&g_mutex);}pthread_mutex_lock(&g_mutex);stop_threads = true;pthread_cond_broadcast(&g_cond); pthread_mutex_unlock(&g_mutex);std::cout << "Master work done " << std::endl;
}void MasterJoin(std::vector<pthread_t> &tids)
{for (auto &tid : tids){pthread_join(tid, nullptr);std::cout << "join success" << std::endl;}
}int main()
{std::vector<pthread_t> tid;MasterStart(tid);Master_awake(tid);MasterJoin(tid);return 0;
}
上述代码简单地演示了条件变量地使用方式，需要注意的是，条件变量一定要在加锁后进行等待。
2、生产消费模型
为了更好地理解条件变量地具体作用，这里介绍一下生产消费模型。
<1>概念的引入
其实我们这个生产消费模型在日常生活中其实挺常见，超市就是典型的生产消费模型。
上图的厂商和消费者我们均可以看成是线程，而超市就是临界资源。这里的商品就是数据，超市实际就是用来存数据品的一段内存空间(数据结构)。这里多个消费者和厂商之间关系，就涉及了多线程的同步和互斥问题。在这个模型中存在着三种关系：生产者和生产者、消费者和消费者(厂商就是生产者)、消费者和生产者、。
<1>生产者和生产者 - 互斥（+同步）
生产者之间的关系必然是互斥的关系。举个生活中的例子，当超市在进货时，不可能会让不同的厂商在同一时间，在同一货架上摆货，这样可能会造成货架上的货物混乱。所以生产者之间一定得是互斥关系。为了避免一个商家一直摆货，我们也可以加上同步关系。
<2>消费者和消费者 - 互斥 （+同步）
消费者之间的关系必然是互斥的。举个生活中的例子，假设现在超市里只有一瓶水，此时有两个想喝水的人同时进入了超市，在这种竞争条件下，两个人都无法获得水。为了避免一个人买完水后，又去拿水结账，一直重复该动作，导致其他人压根无法结账购买货物。我们可以加上同步的关系。
<3>消费者和生产者 - 互斥 + 同步
消费者和生产者之间必须同时拥有这两种关系。这其实很好理解，厂商在上架时，消费者不能去取货物，否则可能会导致货物录入超市系统时出现数目对不上的情况。而厂商不能一直摆货，消费者也不能一直消费货物，这两者的行为必须是同步的。
总结一下，这个超市本质上就是商品的一个缓冲地带。生产消费模型能够为我们提供更好的并发度，将生产和消费的过程进行解耦合。
<2>基于条件变量的生产消费模型的示例代码
在正式编写代码前，我们需要知道的是，了解一下阻塞队列的概念。阻塞队列是一种特殊的队列，当队列为空时，获取队列元素的操作将会被阻塞。当队列满了时，向队列中存放数据的操作也会被阻塞。（这里的阻塞队列就对应着超市）为了实现这种操作，我们就需要应用到条件变量。
<1>单生产单消费模型
ThreadMode.hpp
#ifndef __THREAD_HPP__
#define __THREAD_HPP__#include <iostream>
#include <string>
#include <unistd.h>
#include <functional>
#include <pthread.h>namespace ThreadModule
{template<typename T>using func_t = std::function<void(T)>;// typedef std::function<void(const T&)> func_t;template<typename T>class Thread{public:void Excute(){_func(_data);}public:Thread(func_t<T> func, T data, const std::string name="none-name")//右值: _func(func), _data(data), _threadname(name), _stop(true){}static void *threadroutine(void *args) // 类成员函数，形参是有this指针的！！{Thread<T> *self = static_cast<Thread<T> *>(args);self->Excute();return nullptr;}bool Start(){int n = pthread_create(&_tid, nullptr, threadroutine, this);if(!n){_stop = false;return true;}else{return false;}}void Detach(){if(!_stop){pthread_detach(_tid);}}void Join(){if(!_stop){pthread_join(_tid, nullptr);}}std::string name(){return _threadname;}void Stop(){_stop = true;}T& Data(){return _data;}~Thread() {}private:pthread_t _tid;std::string _threadname;T _data; // 为了让所有的线程访问同一个全局变量func_t<T> _func;bool _stop;};
}#endifBlockQueue.hpp
#include <iostream>
#include <vector>
#include <queue>template <typename T>
class blockqueue
{typedef int data;
private:bool isfull(){return self.size() == capacity;}public:blockqueue(int cap = 20) : capacity(cap){pthread_mutex_init(&_glock, nullptr);pthread_cond_init(&_product_cond, nullptr);pthread_cond_init(&_consume_cond, nullptr);}void Enqueue(const T& in){// 1.加锁pthread_mutex_lock(&_glock);while(isfull()){_product_wait_num++;pthread_cond_wait(&_product_cond, &_glock);//存在伪唤醒情况(消费一个数据，却唤醒了多个生产线程)，可以将外部判断语句变为while_product_wait_num--;}self.push(std::move(in));//1.1唤醒消费线程if(_consume_wait_num > 0)//有消费者在等，才需要唤醒pthread_cond_signal(&_consume_cond);// 2.解锁pthread_mutex_unlock(&_glock);}void Popqueue(T* out){// 1.加锁pthread_mutex_lock(&_glock);while(self.empty())//避免多个线程在被唤醒时，只有一个线程持有锁，其他线程在锁下等待失败，继续向下执行时，出现问题。{_consume_wait_num++;pthread_cond_wait(&_consume_cond, &_glock);//1.让线程进入休眠，当被唤醒后，需要重新持有锁后方能从该位置继续执行。_consume_wait_num--;}*out = self.front();self.pop();//1.1唤醒生产的线程if(_product_wait_num > 0)//有生产者在等，才需要唤醒pthread_cond_signal(&_product_cond);// 2.解锁pthread_mutex_unlock(&_glock);}~blockqueue(){pthread_mutex_destroy(&_glock);pthread_cond_destroy(&_product_cond);pthread_cond_destroy(&_consume_cond);}private:int capacity; // 容量std::queue<T> self;pthread_mutex_t _glock; // 互斥锁pthread_cond_t _product_cond; // 生产者条件变量pthread_cond_t _consume_cond; // 消费者条件变量int _product_wait_num;//等待线程数int _consume_wait_num;};
Main.cc
#include <iostream>
#include <string>
#include <vector>
#include "ThreadMode.hpp"
#include "BlockQueue.hpp"
#include "unistd.h"
using namespace ThreadModule;typedef int data;
using Blockqueue_t = blockqueue<data>*;
pthread_mutex_t _data_lock = PTHREAD_MUTEX_INITIALIZER;data cnt = 100;
void Product(Blockqueue_t ptr)
{while (1){sleep(1);pthread_mutex_lock(&_data_lock);ptr->Enqueue(cnt);std::cout << "product data is :" << cnt << std::endl;cnt--;pthread_mutex_unlock(&_data_lock);}
}
void Consume(Blockqueue_t ptr)
{while (1){sleep(2);data sum;ptr->Popqueue(&sum);std::cout << "comsume data is : " << sum << std::endl;}
}
void StartComm(std::vector<Thread<Blockqueue_t>> &thread, int num, Blockqueue_t ptr, func_t<Blockqueue_t> fun)
{for (int i = 0; i < num; i++){std::string name = "thread - " + std::to_string(i + 1);thread.emplace_back(fun, ptr, name);thread.back().Start();}
}
void StartProducter(std::vector<Thread<Blockqueue_t>> &thread, int num, Blockqueue_t ptr)
{StartComm(thread, num, ptr, Product);
}
void StartConsumer(std::vector<Thread<Blockqueue_t>> &thread, int num, Blockqueue_t ptr)
{StartComm(thread, num, ptr, Consume);
}
void Wait(std::vector<Thread<Blockqueue_t>> &threads)
{for (auto &e : threads){e.Join();}
}
int main()
{blockqueue<data> *ptr = new blockqueue<data>;std::vector<Thread<Blockqueue_t>> threads;StartProducter(threads, 1, ptr);//第二个参数确定生产者数量StartConsumer(threads, 1, ptr);//第二个参数确定消费者数量Wait(threads);return 0;
}运行结果
<3>多生产多消费模型
在实际场景下，单生产单消费的模型其实并不多见，绝大部分是多生产多消费模型。多生产多消费模型的代码其实和上面单消费单生产模型代码一致，我们只需要修改main函数中的参数即可。这里我在阻塞队列中存放的均是int类型的数据，正常情况下，这里面存放的是一个一个的任务。由消费者执行任务，生产者获取任务，这里就不演示具体做法了。
<4>小结
这个模型的优点在于能够提供较好的并发度，虽然临界资源只能由一个线程进行访问，但是拿到任务以后，不同的线程可以并发的处理任务。
3、基于环形队列的生产消费模型
<1>概念引入
上面我们介绍了普通的生产消费模型，在上一个模型中。我们其实可以发现，对于队列的使用(临界资源)其实是不怎么高效的，因为我把队列看成一个整体对数据进行保存。前面我们提到过信号量的概念，该篇文章中，我们提到过临界资源可以被划分称多块使用，也可以被整体使用。既然这里使用的是将临界资源看成整体的方法，那么也应该可以将该资源看成多块进行使用。下面介绍一下基于环形队列的生产消费模型。
<2>具体介绍
环形队列本质上是一个数组(也可以是其他数据结构)，这里不对它的具体实现作描述，简单叙述一下原理即可。环形队列将整体资源分成多块，不用像上面一样进行整体访问，可以做到生产和消费并发执行。
在环形队列中，我们需要使用两个指针，一个指向消费者下标，一个指向生产者下标。在这两个下标指向同一个位置时，可能会存在两种情况：
1、环形队列为满
在这种情况下，我们就要确保消费者先走，不能让生产者先动，否则会造成数据覆盖的现象。
2、环形队列为空
在这种情况下，我们需要确保生产者先走，只有生产者走了，消费者才有东西可以消费。
除开两个下标在同一个位置的情况下，其余情况都是两个指正指向不同的位置。在这种情况下，我们就可以让生产者和消费者并发执行。不过需要注意的是，生产者一定不能超过消费者一圈，并且消费者不能超过生产者，否则会出现问题。
<3>具体实现方法 - 信号量
为了实现上述功能，我们需要引入信号量。前面我们提到信号量就是用来描述临界资源多少的计数器，当信号量大于零时，我们申请临界资源就肯定可以成功，如果等于零，就肯定无法申请成功。这个特性可以让我们省去很多对临界资源的判断。
下面我使用伪代码来简单描述一下环形队列的具体实现过程。
对于生产者来说，空间是比较重要的，而对于消费者来说，数据是比较重要的，而空间和数据都属于资源。所以这里我们需要申请两个信号量，一个用于描述空间资源，一个用于描述数据资源。对于生产者来说，我们首先就要申请空间信号量，由于我们将空间信号量设置成了大于零的初始值，而数据信号量设置成了等于零的初始值。所以在这种情况下，生产者肯定比消费者先走，而消费者只能阻塞在P操作中。在生产者执行完相关代码后，会对数据信号量做V操作，此时消费者申请数据信号量成功，开始向下执行。当执行完所有代码后，会对空间信号量做V操作(这个过程其实就是生产者生产完后，提醒消费者，消费者消费完后提醒生产者，这个过程形成完美闭环)。当两个信号量都不等于0或10时，消费者和生产者就可以同步运行。
<4>基于信号量的环形队列生产消费模型的代码示例
下面对引用上一个模型的代码，简单写一个基于信号量的环形队列生产消费模型。在此之前，我们需要介绍几个信号量的相关接口，方便大家理解
1、sem_init
首先我们需要定义一个sem_t变量，将该变量地址用sem_init进行初始。pshared为零表示，信号量只在线程之间进行共享，如果大于零表示能在进程之间共享。第三个参数表示信号量需要设定的值。
2、sem_destroy
当我们不需要使用信号量时，就可以使用sem_destroy进行销毁。成功返回0，错误返回-1，错误码被设置。
3、sem_wait
sem_wait用于减少信号量，当信号量减少到小于等于零，线程会阻塞。上图第二个接口在出现信号量小于等于零时，不会阻塞，会直接出错返回，第三个接口则是在特定时间内返回。这里我们一般就使用第一个接口，模拟P操作。成功返回0，失败返回-1.
4、sem_post
sem_post用于增加信号量，用于模拟V操作。成功返回0，失败返回-1.
需要注意的是，以上的所有接口，均需链接pthread原生线程库
示例代码：
ThreadMode.hpp（该文件较上文有些许修改）
#ifndef __THREAD_HPP__
#define __THREAD_HPP__#include <iostream>
#include <string>
#include <unistd.h>
#include <functional>
#include <pthread.h>namespace ThreadModule
{template<typename T>using func_t = std::function<void(T,std::string name)>;// typedef std::function<void(const T&)> func_t;template<typename T>class Thread{public:void Excute(){_func(_data,_threadname);}public:Thread(func_t<T> func, T data, const std::string name="none-name")//右值: _func(func), _data(data), _threadname(name), _stop(true){}static void *threadroutine(void *args) // 类成员函数，形参是有this指针的！！{Thread<T> *self = static_cast<Thread<T> *>(args);self->Excute();return nullptr;}bool Start(){int n = pthread_create(&_tid, nullptr, threadroutine, this);if(!n){_stop = false;return true;}else{return false;}}void Detach(){if(!_stop){pthread_detach(_tid);}}void Join(){if(!_stop){pthread_join(_tid, nullptr);}}std::string name(){return _threadname;}void Stop(){_stop = true;}T& Data(){return _data;}~Thread() {}private:pthread_t _tid;std::string _threadname;T _data; // 为了让所有的线程访问同一个全局变量func_t<T> _func;bool _stop;};
}#endifRingQueue.hpp
#include <vector>
#include <iostream>
#include <string>
#include <semaphore.h>template <class T>
class RingQueue
{
private:void P(sem_t &sem){sem_wait(&sem);}void V(sem_t &sem){sem_post(&sem);}void Lock(pthread_mutex_t &lock){pthread_mutex_lock(&lock);}void Unlock(pthread_mutex_t &lock){pthread_mutex_unlock(&lock);}public:RingQueue(int cap = 10) : _cap(cap)//初始值默认设置成10{pthread_mutex_init(&_productor_mutex, nullptr);pthread_mutex_init(&_consumer_mutex, nullptr);sem_init(&room, 0, 10);sem_init(&data, 0, 0);_product_index = 0;_consume_index = 0;queue.resize(cap);}void Enqueue(const T &date){//申请信号量P(room);//由于PV操作本身就是原子的，所以这里是不用加锁//加锁，保护临界资源Lock(_productor_mutex);queue[_product_index++] = date;_product_index %= _cap;Unlock(_productor_mutex);V(data);}void Popqueue(T *date){P(data);Lock(_consumer_mutex);*date = queue[_consume_index];queue[_consume_index++] = T();(_consume_index) %= _cap;Unlock(_consumer_mutex);V(room);}~RingQueue(){sem_destroy(&room);sem_destroy(&data);pthread_mutex_destroy(&_productor_mutex);pthread_mutex_destroy(&_consumer_mutex);}private:int _cap;std::vector<T> queue;// 空间信号量sem_t room;sem_t data;// 空间信号量下标int _product_index;int _consume_index;// 多线程需要加锁pthread_mutex_t _productor_mutex;pthread_mutex_t _consumer_mutex;
};Main.cc（较上个示例代码有一定改动）
#include <iostream>
#include <string>
#include <vector>
#include "ThreadMode.hpp"
#include "unistd.h"
#include "RingQueue.hpp"
using namespace ThreadModule;typedef int data;
using Ringqueue_t = RingQueue<data>*;
pthread_mutex_t _data_lock = PTHREAD_MUTEX_INITIALIZER;data cnt = 100;
void Product(Ringqueue_t ptr,std::string name)
{while (1){sleep(1);pthread_mutex_lock(&_data_lock);ptr->Enqueue(cnt);std::cout << "product data is :" << cnt << "---[" << name << "]" << std::endl;cnt--;pthread_mutex_unlock(&_data_lock);}
}
void Consume(Ringqueue_t ptr,std::string name)
{while (1){sleep(2);data sum;ptr->Popqueue(&sum);std::cout << "comsume data is : " << sum << "---[" << name << "]" << std::endl;}
}
void StartComm(std::vector<Thread<Ringqueue_t>> &thread, int num, Ringqueue_t ptr, func_t<Ringqueue_t> fun,const std::string cname)
{for (int i = 0; i < num; i++){std::string name = "thread-" + std::to_string(i + 1) + " + " + cname;thread.emplace_back(fun, ptr, name);}
}
void StartProducter(std::vector<Thread<Ringqueue_t>> &thread, int num, Ringqueue_t ptr)
{StartComm(thread, num, ptr, Product, "Productor");
}
void StartConsumer(std::vector<Thread<Ringqueue_t>> &thread, int num, Ringqueue_t ptr)
{StartComm(thread, num, ptr, Consume,"Consumer");
}
void Wait(std::vector<Thread<Ringqueue_t>> &threads)
{for (auto &e : threads){e.Join();}
}
void StartAll(std::vector<Thread<Ringqueue_t>>& thread)
{for(auto& e : thread){e.Start();}
}
int main()
{RingQueue<data> *ptr = new RingQueue<data>;std::vector<Thread<Ringqueue_t>> threads;StartProducter(threads, 4, ptr);StartConsumer(threads, 5, ptr);StartAll(threads);Wait(threads);return 0;
}运行结果：
以上就是所有内容，文中如有不对之处，还望各位大佬指正，谢谢！！！
