阅读量:2
引言
今天，我们继续学习Linux线程本分，在Linux条件变量中，我们对条件变量的做了详细的说明，今天我们要利用条件变量来引出我们的另一个话题——信号量内容的学习。
1.复习条件变量
在上一期博客中，我们没有对条件变量做具体的使用，所以，这里我们通过一份代码来复习一下，接下来，我们实现基于BlockingQueue的生产者消费者模型。
1.1何为基于BlockingQueue的生产者消费者模型
BlockingQueue在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)
如图：
1.2分析该模型
这里我想写多个生产线程和多个消费线程的模型
我们来分析一下。
- 首先生产任务的过程和消费任务的过程必须是互斥关系，不可以同时访问该队列（此时，这个队列是共享资源）。
- 当队列满时，生产线程就不能再生产任务，必须在特定的条件变量下等待；同理当队列为空时，消费线程就不能再消费任务，也必须在特定的条件变量下等待。
所以，类应这样设计：
template<class T> class BlockQueue { public: BlockQueue(const int &maxcap=gmaxcap):_maxcap(maxcap) { pthread_mutex_init(&_mutex,nullptr); pthread_cond_init(&_pcond,nullptr); pthread_cond_init(&_ccond,nullptr); } void push(const T&in)//输入型参数，const & { pthread_mutex_lock(&_mutex); while(is_full()) { pthread_cond_wait(&_pcond,&_mutex); } _q.push(in); pthread_cond_signal(&_ccond); pthread_mutex_unlock(&_mutex); } void pop(T*out) { pthread_mutex_lock(&_mutex); while(is_empty()) { pthread_cond_wait(&_ccond,&_mutex); } *out=_q.front(); _q.pop(); pthread_cond_signal(&_pcond); pthread_mutex_unlock(&_mutex); } ~BlockQueue() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_ccond); pthread_cond_destroy(&_pcond); } private: bool is_empty() { return _q.empty(); } bool is_full() { return _q.size()==_maxcap; } private: std::queue<T> _q; int _maxcap; //队列中元素的上线 pthread_mutex_t _mutex; pthread_cond_t _pcond; //生产者对应的条件变量 pthread_cond_t _ccond; }; 由于我们不知道存储的数据类型，所以这里我们选择使用泛型编程的方式。
接下来就是要生产任务，为了可以观察到整个生产和消费任务的过程，我们可以生成两个随机数，然后进行运算。代码如下：
class CalTask { using func_t = function<int(int, int, char)>; public: CalTask() {} CalTask(int x, int y, char op, func_t func) :_x(x),_y(y),_op(op),_callback(func) {} string operator()() { int result=_callback(_x,_y,_op); char buffer[1024]; snprintf(buffer,sizeof buffer,"%d %c %d=%d",_x,_op,_y,result); return buffer; } string toTaskstring() { char buffer[1024]; snprintf(buffer,sizeof buffer,"%d %c %d=?",_x,_op,_y); return buffer; } private: int _x; int _y; char _op; func_t _callback; }; const char*oper="+-*/%"; int mymath(int x,int y,char op) { int result=0; switch(op) { case '+': result=x+y; break; case '-': result=x-y; break; case '*': result=x*y; break; case '/': if(y==0) { cerr<<"div zero error"<<endl; result=-1; } else { result=x/y; } break; case '%': if(y==0) { cerr<<"mod zero error"<<endl; result=-1; } else { result=x%y; } default: break; } return result; } 接下来，我们来写整体的代码。
1.3完整代码
我们要创建三个文件：BlockQueue.hpp Task.hpp Main.cc各文件内容如下所示：
BlockQueue.hpp
#pragma once #include<iostream> #include<pthread.h> #include<cstring> #include<unistd.h> #include<cassert> #include<queue> using namespace std; const int gmaxcap=100; template<class T> class BlockQueue { public: BlockQueue(const int &maxcap=gmaxcap):_maxcap(maxcap) { pthread_mutex_init(&_mutex,nullptr); pthread_cond_init(&_pcond,nullptr); pthread_cond_init(&_ccond,nullptr); } void push(const T&in)//输入型参数，const & { pthread_mutex_lock(&_mutex); while(is_full()) { pthread_cond_wait(&_pcond,&_mutex); } _q.push(in); pthread_cond_signal(&_ccond); pthread_mutex_unlock(&_mutex); } void pop(T*out) { pthread_mutex_lock(&_mutex); while(is_empty()) { pthread_cond_wait(&_ccond,&_mutex); } *out=_q.front(); _q.pop(); pthread_cond_signal(&_pcond); pthread_mutex_unlock(&_mutex); } ~BlockQueue() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_ccond); pthread_cond_destroy(&_pcond); } private: bool is_empty() { return _q.empty(); } bool is_full() { return _q.size()==_maxcap; } private: std::queue<T> _q; int _maxcap; //队列中元素的上线 pthread_mutex_t _mutex; pthread_cond_t _pcond; //生产者对应的条件变量 pthread_cond_t _ccond; }; Task.hpp
#pragma once #include <iostream> #include <string> #include <cstdio> #include<string> #include <functional> using namespace std; class CalTask { using func_t = function<int(int, int, char)>; public: CalTask() {} CalTask(int x, int y, char op, func_t func) :_x(x),_y(y),_op(op),_callback(func) {} string operator()() { int result=_callback(_x,_y,_op); char buffer[1024]; snprintf(buffer,sizeof buffer,"%d %c %d=%d",_x,_op,_y,result); return buffer; } string toTaskstring() { char buffer[1024]; snprintf(buffer,sizeof buffer,"%d %c %d=?",_x,_op,_y); return buffer; } private: int _x; int _y; char _op; func_t _callback; }; const char*oper="+-*/%"; int mymath(int x,int y,char op) { int result=0; switch(op) { case '+': result=x+y; break; case '-': result=x-y; break; case '*': result=x*y; break; case '/': if(y==0) { cerr<<"div zero error"<<endl; result=-1; } else result=x/y; } break; case '%': if(y==0) { cerr<<"mod zero error"<<endl; result=-1; } else { result=x%y; } default: break; } return result; } Main.cc
include "BlockQueue.hpp" #include "Task.hpp" #include<sys/types.h> #include<unistd.h> #include<ctime> #include<iostream> using namespace std; void *productor(void *bqs_) { BlockQueue<CalTask> *bqs=static_cast<BlockQueue<CalTask>*>(bqs_); while(true) { int x=rand()%10+1; int y=rand()%5+1; int opercode=rand()%(sizeof(oper)); CalTask T(x,y,oper[opercode],mymath); bqs->push(T); cout<<"生产任务: "; cout<<T.toTaskstring()<<endl; sleep(1); } } void *consumer(void *bqs_) { BlockQueue<CalTask>*bqs=static_cast<BlockQueue<CalTask>*>(bqs_); while(true) { CalTask T; bqs->pop(&T); cout<<"消费任务: "; cout<<T()<<endl; } } int main() { BlockQueue<CalTask> bqs; pthread_t p[5]; pthread_t c[5]; for(int i=0;i<5;i++) { pthread_create(&p[i],nullptr,productor,&bqs); pthread_create(&c[i],nullptr,consumer,&bqs); } for(int i=0;i<5;i++) { pthread_join(p[i],nullptr); pthread_join(c[i],nullptr); } } 在代码中，有几个点需要注意一下：
第一点：
pthread_cond_wait的第二个参数一定是我们正在使用的互斥锁，这个函数在被运行时，会以原子性的方式将锁释放，然后将自己挂起，等待被条件变量唤醒。该函数在被唤醒时，会自动重新获取持有的锁，然后继续向下执行。
假如数个生产者线程一起被唤醒，然后先后持有锁，接着继续生产任务，当队列剩余的空间小于这些生产者生产的任务时，就会出现问题，所以让所有被唤醒的线程先通过while循环，如果有剩余的空间，再进行任务的生产活动。
生产线程这样处理，消费线程也要这样处理
大家可以在自己试这敲一下，有问题可以在评论区和我交流。
接下来，我们来查找一下这些代码有哪些"不足的地方"
2.代码中的“不足”
一个线程在操作临界资源时，临界资源必须是满足条件的，然后线程才能对临界资源进行操作。比如：在如上代码中，生产者线程只有在队列(临界资源）有剩余空间的条件下，才能进行下一步操作。
可是，临界资源是否满足生产和消费的条件，我们不能事前得知，只等进入临界资源后，再进行进一步的检测。
所以，一般访问临界资源的过程为：先加锁，再检测，如果条件满足，就进行下一步的操作；反之，就将该线程挂起，释放锁，然后挂起等待，等到条件满足时，重新获得锁，接着进行下一步操作。
因为不可能事先得知是否满足条件，所以我们只能先加锁，进入临界资源内部进行检测。
只要我们申请了信号量，就默认对这部分资源的整体使用，但通常情况下，我们使用的仅仅是临界资源的一小部分。
实际情况中，有没有可能不同的线程访问临界资源不同部分的情况，有可能。所以，前辈大佬们给出了一种解决方案——信号量。
3.信号量
3.1什么是信号量
信号量的本质是一把计数器，一把衡量临界资源多少的计数器。只要拥有信号量，就在未来一定能够拥有临界资源的一部分。
申请信号量的本质：就是对临界资源的预定机制。
比如：我想去看电影，首先我要买票。我一旦买到票，无论我去不去看电影，都会有一个位置属于我。买票的过程==申请信号信号量的过程。
所以，在访问临界资源之前，我们可以申请信号量。通过申请信号量，我们就可以获知临界资源的使用情况。①只要申请成功，就一定有我可以访问的资源。②只要申请失败，说明条件不就绪，只能等待。如此，就不需要进入临界资源再进行检测了。
3.2信号量的相关接口
如上这些借口如果调用成功的话，返回0；调用失败的话，返回-1，并且错误原因被设置。
我们知道信号量的本质是一把计数器，所以信号量必须可以进行递增和递减的操作。
- 信号量-1：申请资源，其过程必须是原子性的。简称P操作。
- 信号量+1：归还资源，其过程必须是原子性的。简称V操作。
所以，信号量的核心操作：PV原语。
接下来，我们就使用信号量来完成我们的基于环形队列的生产消费模型。
3.3用信号量来实现基于环形队列的生产消费模型
3.3.1对环形队列的简单介绍
相信大家在C++学习期间到都模拟实现过环形队列队列。如图：
环形队列的逻辑结构为环形，但其存储结构实际上就是队列，其实就是一个数组，只不过用下标不断的%上队列的长度。
大家在模拟实现环形队列时，大家必定遇到的问题是：当rear==front时，究竟是环形队列已满还是环形队列为空呢？其实，这个问题有多种处理方式，今天就不讲了。
今天，我们的基于环形队列的生产消费模型必须遵守哪些规则呢？
我们来讲一个故事：
张三和李四在一个房间里做游戏，这个房间里有一张大圆桌，桌子上有很多的盘子。规定张三往每个盘子里放一个桃子🍑，然后李四在后边吃桃子🍑，由于李四还要吃桃子，所以速度一定比张三放的速度满。
总结一下，我们发现张三和李四必须满足这些规律：
- 李四不可以超过张三——消费者不可以超过生产者。
- 张三不可以把李四套一个圈——生产者不可以把消费者套一个圈。
- 张三和李四什么时候在一起？①盘子全为空，张三和李四在一起，张三先运行（生产者先运行）。②盘子全为满，张三和李四在一起，李四先运行（消费者先运行）。③其他情况，张三和李四指向不同的位置。
我们将这些规则迁移到环形队列的生产消费模型，就是生产消费模型应该遵守的规则：
①消费者不能超过生产者。②生产者不能把消费者套一个圈。③生产者和消费者什么情况下会在一起呢？空的时候和满的时候，对应不同的处理方式。④只要生产者和消费者指向不同的位置，就可以实现生产者和消费者的并发执行。只有在为空和为 满时，才会出现同步和互斥问题。
那这些规则由什么来保证呢？信号量。信号量是表征临界资源中资源数目的。
1.对于生产者而言，看中的是队列中的剩余空间——空间资源定义一个信号量。
2.对于消费者而言，看中的是队列中的数据——数据资源定义一个信号量。
接下来，我们基于这份伪代码来理解一下，看看能否满足我们的规则。
生产者关注的是队列里的剩余空间，在队列为空时剩余空间为10，所以生产者可以顺利申请到信号量。但是由于空间中这部分资源已经被占用，所以无法归还。但是消费者所关注的队列中的数据资源不知不觉中已经多了一份。所以对消费者信号量应进行V操作。
消费者关注的是队列中的数据资源，队列刚开始为空时，数据资源为0，消费者申请失败。等到生产者申请神域空间成功后，生产了数据。所以消费者可以成功申请到数据资源信号量，然后消费数据。但不知不觉，队列中的剩余空间多了一份，所以应对剩余空间资源的信号量进行V操作。
若队列满时，剩余空间信号量为0，生产者申请信号量失败。此时，数据资源信号量为满，消费者可以申请到信号量，从而进行操作。所以必须消费者先运行。
若队列空时，数据资源信号量为0，消费者申请信号量失败。此时，剩余空间信号量为满，生产者可以申请到信号量，从而进行操作。所以必须生产者先运行。
所以，这伪代码完全符合我们的规则。接下来，我们编写单生产进程和单消费进程的代码。
编写代码
我们创建三个源文件：RingQueue.hpp main.cc Task.hpp
Ringqueue.hpp:
#pragma once #include <iostream> #include <vector> #include <cassert> #include <semaphore.h> #include <pthread.h> static const int gcap = 5; template<class T> class RingQueue { private: void P(sem_t &sem) { int n = sem_wait(&sem); assert(n == 0); // if (void)n; } void V(sem_t &sem) { int n = sem_post(&sem); assert(n == 0); (void)n; } public: RingQueue(const int &cap = gcap): _queue(cap), _cap(cap) { int n = sem_init(&_spaceSem, 0, _cap); assert(n == 0); n = sem_init(&_dataSem, 0, 0); assert(n == 0); _productorStep = _consumerStep = 0; pthread_mutex_init(&_pmutex, nullptr); pthread_mutex_init(&_cmutex, nullptr); } // 生产者 void Push(const T &in) { // ?: 这个代码 有没有优化的可能 // 你认为：现加锁，后申请信号量，还是现申请信号量，在加锁？ P(_spaceSem); // 申请到了空间信号量，意味着，我一定能进行正常的生产 pthread_mutex_lock(&_pmutex); _queue[_productorStep++] = in; _productorStep %= _cap; pthread_mutex_unlock(&_pmutex); V(_dataSem); } // 消费者 void Pop(T *out) { // 你认为：现加锁，后申请信号量，还是现申请信号量，在加锁？ P(_dataSem); pthread_mutex_lock(&_cmutex); *out = _queue[_consumerStep++]; _consumerStep %= _cap; pthread_mutex_unlock(&_cmutex); V(_spaceSem); } ~RingQueue() { sem_destroy(&_spaceSem); sem_destroy(&_dataSem); pthread_mutex_destroy(&_pmutex); pthread_mutex_destroy(&_cmutex); } private: std::vector<T> _queue; int _cap; sem_t _spaceSem; // 生产者 想生产，看中的是什么资源呢? 空间资源 sem_t _dataSem; // 消费者 想消费，看中的是什么资源呢? 数据资源 int _productorStep; int _consumerStep; pthread_mutex_t _pmutex; pthread_mutex_t _cmutex; }; Task.hpp
#pragma once #include <iostream> #include <string> #include <cstdio> #include <functional> class Task { using func_t = std::function<int(int,int,char)>; // typedef std::function<int(int,int)> func_t; public: Task() {} Task(int x, int y, char op, func_t func) :_x(x), _y(y), _op(op), _callback(func) {} std::string operator()() { int result = _callback(_x, _y, _op); char buffer[1024]; snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result); return buffer; } std::string toTaskString() { char buffer[1024]; snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y); return buffer; } private: int _x; int _y; char _op; func_t _callback; }; const std::string oper = "+-*/%"; int mymath(int x, int y, char op) { 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::cerr << "div zero error!" << std::endl; result = -1; } else result = x / y; } break; case '%': { if (y == 0) { std::cerr << "mod zero error!" << std::endl; result = -1; } else result = x % y; } break; default: // do nothing break; } return result; } main.cc
#include "RingQueue.hpp" #include "Task.hpp" #include <pthread.h> #include <ctime> #include <cstdlib> #include <sys/types.h> #include <unistd.h> std::string SelfName() { char name[128]; snprintf(name, sizeof(name), "thread[0x%x]", pthread_self()); return name; } void *ProductorRoutine(void *rq) { // RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq); RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq); while(true) { // version1 // int data = rand() % 10 + 1; // ringqueue->Push(data); // std::cout << "生产完成，生产的数据是：" << data << std::endl; // version2 // 构建or获取任务 --- 这个是要花时间的！ int x = rand() % 10; int y = rand() % 5; char op = oper[rand()%oper.size()]; Task t(x, y, op, mymath); // 生产任务 ringqueue->Push(t); // 输出提示 std::cout << SelfName() << ", 生产者派发了一个任务: " << t.toTaskString() << std::endl; // sleep(1); } } void *ConsumerRoutine(void *rq) { // RingQueue<int> *ringqueue = static_cast<RingQueue<int> *>(rq); RingQueue<Task> *ringqueue = static_cast<RingQueue<Task> *>(rq); while(true) { //version1 // int data; // ringqueue->Pop(&data); // std::cout << "消费完成，消费的数据是：" << data << std::endl; // sleep(1); // version2 Task t; //消费任务 ringqueue->Pop(&t); std::string result = t(); // 消费也是要花时间的！ std::cout << SelfName() << ", 消费者消费了一个任务: " << result << std::endl; // sleep(1); } } int main() { srand((unsigned int)time(nullptr) ^ getpid() ^ pthread_self() ^ 0x71727374); // RingQueue<int> *rq = new RingQueue<int>(); RingQueue<Task> *rq = new RingQueue<Task>(); // 单生产，单消费，多生产，多消费 --> 只要保证，最终进入临界区的是一个生产，一个消费就行！ // 多生产，多消费的意义？？ pthread_t p[4], c[8]; for(int i = 0; i < 4; i++) pthread_create(p+i, nullptr, ProductorRoutine, rq); for(int i = 0; i < 8; i++) pthread_create(c+i, nullptr, ConsumerRoutine, rq); for(int i = 0; i < 4; i++) pthread_join(p[i], nullptr); for(int i = 0; i < 8; i++) pthread_join(c[i], nullptr); delete rq; return 0; } 大家可以自己敲一敲，试一下。
写到这里，这篇博客就结束了，下篇博客我们再见。
