政务网站建设目的 意义,可以自己做免费网站吗,东莞市手机网站建设多少钱,维护网站建设欢迎来到 破晓的历程的 博客 引言
今天#xff0c;我们继续学习Linux线程本分#xff0c;在Linux条件变量中#xff0c;我们对条件变量的做了详细的说明#xff0c;今天我们要利用条件变量来引出我们的另一个话题——信号量内容的学习。
1.复习条件变量
在上一期博客中我们继续学习Linux线程本分在Linux条件变量中我们对条件变量的做了详细的说明今天我们要利用条件变量来引出我们的另一个话题——信号量内容的学习。
1.复习条件变量
在上一期博客中我们没有对条件变量做具体的使用所以这里我们通过一份代码来复习一下接下来我们实现基于BlockingQueue的生产者消费者模型。
1.1何为基于BlockingQueue的生产者消费者模型
BlockingQueue在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放入了元素当队列满时往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的线程在对阻塞队列进程操作时会被阻塞) 如图
1.2分析该模型
这里我想写多个生产线程和多个消费线程的模型 我们来分析一下。
首先生产任务的过程和消费任务的过程必须是互斥关系不可以同时访问该队列此时这个队列是共享资源。当队列满时生产线程就不能再生产任务必须在特定的条件变量下等待同理当队列为空时消费线程就不能再消费任务也必须在特定的条件变量下等待。 所以类应这样设计
templateclass T
class BlockQueue
{
public:BlockQueue(const int maxcapgmaxcap):_maxcap(maxcap){pthread_mutex_init(_mutex,nullptr);pthread_cond_init(_pcond,nullptr);pthread_cond_init(_ccond,nullptr);}void push(const Tin)//输入型参数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::queueT _q;int _maxcap; //队列中元素的上线pthread_mutex_t _mutex;pthread_cond_t _pcond; //生产者对应的条件变量pthread_cond_t _ccond;
};由于我们不知道存储的数据类型所以这里我们选择使用泛型编程的方式。 接下来就是要生产任务为了可以观察到整个生产和消费任务的过程我们可以生成两个随机数然后进行运算。代码如下
class CalTask
{using func_t functionint(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 result0;switch(op){case :resultxy;break;case -:resultx-y;break;case *:resultx*y;break;case /:if(y0){cerrdiv zero errorendl;result-1;}else{resultx/y;}break;case %:if(y0){cerrmod zero errorendl;result-1;}else{resultx%y;}default:break;}return result;
}
接下来我们来写整体的代码。
1.3完整代码
我们要创建三个文件BlockQueue.hpp Task.hpp Main.cc各文件内容如下所示
BlockQueue.hpp
#pragma once
#includeiostream
#includepthread.h
#includecstring
#includeunistd.h
#includecassert
#includequeue
using namespace std;
const int gmaxcap100;
templateclass T
class BlockQueue
{
public:BlockQueue(const int maxcapgmaxcap):_maxcap(maxcap){pthread_mutex_init(_mutex,nullptr);pthread_cond_init(_pcond,nullptr);pthread_cond_init(_ccond,nullptr);}void push(const Tin)//输入型参数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::queueT _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
#includestring
#include functional
using namespace std;
class CalTask
{using func_t functionint(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 result0;switch(op){case :resultxy;break;case -:resultx-y;break;case *:resultx*y;break;case /:if(y0){cerrdiv zero errorendl;result-1;}elseresultx/y;}break;case %:if(y0){cerrmod zero errorendl;result-1;}else{resultx%y;}default:break;}return result;
}Main.cc
include BlockQueue.hpp
#include Task.hpp
#includesys/types.h
#includeunistd.h
#includectime
#includeiostream
using namespace std;void *productor(void *bqs_)
{BlockQueueCalTask *bqsstatic_castBlockQueueCalTask*(bqs_);while(true){int xrand()%101;int yrand()%51;int opercoderand()%(sizeof(oper));CalTask T(x,y,oper[opercode],mymath);bqs-push(T);cout生产任务: ;coutT.toTaskstring()endl;sleep(1);}
}
void *consumer(void *bqs_)
{BlockQueueCalTask*bqsstatic_castBlockQueueCalTask*(bqs_);while(true){CalTask T;bqs-pop(T);cout消费任务: ;coutT()endl;}
}
int main()
{BlockQueueCalTask bqs;pthread_t p[5];pthread_t c[5];for(int i0;i5;i){pthread_create(p[i],nullptr,productor,bqs);pthread_create(c[i],nullptr,consumer,bqs);}for(int i0;i5;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学习期间到都模拟实现过环形队列队列。如图 环形队列的逻辑结构为环形但其存储结构实际上就是队列其实就是一个数组只不过用下标不断的%上队列的长度。 大家在模拟实现环形队列时大家必定遇到的问题是当rearfront时究竟是环形队列已满还是环形队列为空呢其实这个问题有多种处理方式今天就不讲了。 今天我们的基于环形队列的生产消费模型必须遵守哪些规则呢 我们来讲一个故事 张三和李四在一个房间里做游戏这个房间里有一张大圆桌桌子上有很多的盘子。规定张三往每个盘子里放一个桃子然后李四在后边吃桃子由于李四还要吃桃子所以速度一定比张三放的速度满。 总结一下我们发现张三和李四必须满足这些规律
李四不可以超过张三——消费者不可以超过生产者。张三不可以把李四套一个圈——生产者不可以把消费者套一个圈。张三和李四什么时候在一起①盘子全为空张三和李四在一起张三先运行生产者先运行。②盘子全为满张三和李四在一起李四先运行消费者先运行。③其他情况张三和李四指向不同的位置。
我们将这些规则迁移到环形队列的生产消费模型就是生产消费模型应该遵守的规则 ①消费者不能超过生产者。②生产者不能把消费者套一个圈。③生产者和消费者什么情况下会在一起呢空的时候和满的时候对应不同的处理方式。④只要生产者和消费者指向不同的位置就可以实现生产者和消费者的并发执行。只有在为空和为 满时才会出现同步和互斥问题。 那这些规则由什么来保证呢信号量。信号量是表征临界资源中资源数目的。 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.hstatic const int gcap 5;templateclass 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::vectorT _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 functionalclass Task
{using func_t std::functionint(int,int,char);// typedef std::functionint(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;}elseresult x / y;}break;case %:{if (y 0){std::cerr mod zero error! std::endl;result -1;}elseresult x % y;}break;default:// do nothingbreak;}return result;
}main.cc
#include RingQueue.hpp
#include Task.hpp
#include pthread.h
#include ctime
#include cstdlib
#include sys/types.h
#include unistd.hstd::string SelfName()
{char name[128];snprintf(name, sizeof(name), thread[0x%x], pthread_self());return name;
}void *ProductorRoutine(void *rq)
{// RingQueueint *ringqueue static_castRingQueueint *(rq);RingQueueTask *ringqueue static_castRingQueueTask *(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)
{// RingQueueint *ringqueue static_castRingQueueint *(rq);RingQueueTask *ringqueue static_castRingQueueTask *(rq);while(true){//version1// int data;// ringqueue-Pop(data);// std::cout 消费完成消费的数据是 data std::endl;// sleep(1);// version2Task 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);// RingQueueint *rq new RingQueueint();RingQueueTask *rq new RingQueueTask();// 单生产单消费多生产多消费 -- 只要保证最终进入临界区的是一个生产一个消费就行// 多生产多消费的意义pthread_t p[4], c[8];for(int i 0; i 4; i) pthread_create(pi, nullptr, ProductorRoutine, rq);for(int i 0; i 8; i) pthread_create(ci, 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;
}大家可以自己敲一敲试一下。 写到这里这篇博客就结束了下篇博客我们再见。
