文章目录
- 信号量概念
- 信号量接口
- 初始化
- 销毁
- 等待
- 发布
- 基于环形队列的生产者消费者模型
- 编码
- Common.h
- LockGuard.hpp
- Task.hpp
- sem.hpp
- RingQueue.hpp
- ConProd.cc
信号量概念
POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。
什么是信号量？
共享资源—>保证任何时刻都只有一个执行流进行访问----->也就有了之前的临界资源，临界区的概念
之前的互斥加锁—>将共享资源当做整个使用。
但是不同的线程可能访问同一个共享资源的不同区域，这样如何我们加个整体锁必定带来整个进行效率的降低。
如果一个共享资源不当做一个整体，而让不同的执行流访问不同的区域的话，那么不就提高了并发度吗？是的。
而只有当不同的线程访问同一个区域的时候我们再进行同步与互斥。
两个问题
1.我们怎么知道一整个共享区，一共有多少个资源，还剩多少个资源？
2.我们怎么保证这个资源就是给我们的？（程序员编码保证）我们怎么知道线程一定可以具有一个共享资源（信号量保证）
电影院买票的例子：
买票的本质：叫做资源（电影院座位）的预订机制
信号量本质：是一个计数器，访问临界资源的时候，必须先申请资源（sem–预订资源），使用完毕后信号量资源必须释放（sem++）
如何理解信号量的使用—>我们申请一个信号量---->当前执行流一定具有一个资源，可以被它使用—>是哪个资源呢？---->需要程序员结合场景，自定义编码完成。
信号量接口
初始化
int sem_init(sem_t *sem, int pshared, unsigned int value);
//参数：
//pshared:0表示线程间共享，非零表示进程间共享
//value：信号量初始值
销毁
int sem_destroy(sem_t *sem);
等待
//功能：等待信号量，会将信号量的值减1
int sem_wait(sem_t *sem);
发布
//功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。
int sem_post(sem_t *sem);
基于环形队列的生产者消费者模型
环形队列采用数组模拟，用模运算来模拟环状特性环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态
但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程
什么时候生产者和消费者会访问同一个位置？
1.环形队列空的时候，生产者和消费者会访问同一个位置
2.环形队列满的时候，生产者和消费者会访问同一个位置
3.其它情况不会访问同一个位置。
那么该如何解决？
1.用计数器解决
2.专门浪费一个格子解决
如果生产者和消费者指向了环形结构的同一个位置；那么生产者和消费者要有互斥或者同步的问题，不然生产者和消费者指向的都是不同的位置。
当生产者和消费者指向同一个位置，具有互斥同步关系就可以。
当生产者和消费者不指向同一个位置想让他们并发执行呢？
期望：
1.生产者不能将消费者套圈。
2.消费者超过生产者
3.环形队列为空：一定要让生产者先行
4.环形队列为满：一定要让消费者先行
怎么编码保证？
生产者：最关注的是空间资源
消费者：最关注的是空间里面的数据资源
最开始计数器值：
生产者：spaceSem---->为最大的N
消费者：dataSem---->为0
生产者：生产一个数据
P(spaceSem)—>spaceSem- -
V(dataSem)----->dataSem++
消费者：消费一个数据
P(dataSem)—>dataSem- -
V(spaceSem)----->spaceSem++
编码
Common.h
#pragma once
#include <iostream>
#include <array>
#include <functional>
#include <ctime>
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>#define RingQueueNum 5//环形队列空间大小
#define CONSUMRT_NUM 2//生产者个数
#define PRODUCTER_NUM 2//消费者个数
//生产者和消费者个数可以自定义
LockGuard.hpp
#pragma once
#include "Common.h"//自己封装的锁，可以自动初始化和销毁
class Mutex
{
public:Mutex(){pthread_mutex_init(&_mtx,nullptr);}pthread_mutex_t& GetMutex(){return _mtx;}~Mutex(){pthread_mutex_destroy(&_mtx);}
private:pthread_mutex_t _mtx;
};//自己封装RAII风格的lock和unlock，会自动解锁解锁
class LockGuard
{
public:LockGuard(Mutex* mtx):_mtx(mtx){pthread_mutex_lock(&_mtx->GetMutex());//std::cout << "lock" << std::endl;}~LockGuard(){pthread_mutex_unlock(&_mtx->GetMutex());//std::cout << "unlock" << std::endl;}
private:Mutex* _mtx;
};
Task.hpp
#pragma once
#include "Common.h"typedef std::function<int(int, int)> func_t;//自己写的任务
class Task
{
public:Task(){}Task(int x, int y, func_t func): _x(x),_y(y),_func(func){}int operator()(){return _func(_x, _y);}public:int _x;int _y;func_t _func;
};
sem.hpp
#pragma once
#include "Common.h"//自己封装一个信号量P(申请信号量)，V(释放信号量)
class Sem
{
public:Sem(int value = RingQueueNum){sem_init(&_sem,0,value);}void P(){sem_wait(&_sem);}void V(){sem_post(&_sem);}~Sem(){sem_destroy(&_sem);}
private:sem_t _sem;
};
RingQueue.hpp
#pragma once
#include "Common.h"
#include "LockGuard.hpp"
#include "Sem.hpp"template<class T>
class RingQueue
{
public:RingQueue():_space_sem(RingQueueNum),_data_sem(0){}void Push(T& in){//先申请信号量_space_sem.P();LockGuard lock(&_producter_mtx);_rq[product_index++] = in;product_index %= RingQueueNum;_data_sem.V();}void Pop(T* out){//先申请信号量_data_sem.P();LockGuard lock(&_cosumer_mtx);*out = _rq[consumer_index++];consumer_index %= RingQueueNum;_space_sem.V();}
private:std::array<T,RingQueueNum> _rq;//环形队列，本质是数组int consumer_index = 0;//消费者数组下标int product_index = 0;//生产者数组下标Mutex _cosumer_mtx;//消费者锁Mutex _producter_mtx;//生产者锁Sem _space_sem;//空间资源信号量Sem _data_sem;//数据资源信号量
};
ConProd.cc
#include "Common.h"
#include "RingQueue.hpp"
#include "Task.hpp"int myadd(int x, int y)
{return x + y;
}
void *ConsumerRoutine(void *args)
{RingQueue<Task> *rq = (RingQueue<Task> *)args;while (true){//获取任务Task t;rq->Pop(&t);//执行任务std::cout << pthread_self() << "|Get a Task: " << t._x << " + " << t._y << "=" << t() << std::endl;sleep(1);}
}
void *ProducterRoutine(void *args)
{RingQueue<Task> *rq = (RingQueue<Task> *)args;while (true){//制作任务int x = rand() % 100 + 1;int y = rand() % 100 + 1;Task t(x, y, myadd);std::cout << pthread_self() << "|Make a Task: " << x << " + " << y << "=?" << std::endl;//发送任务rq->Push(t);sleep(1);}
}
int main()
{srand((unsigned int)time(nullptr) ^ getpid() ^ 0x77777);//定义线程pthread_t consumer[CONSUMRT_NUM];pthread_t producter[PRODUCTER_NUM];//定义环形队列RingQueue<Task> *ring_queue = new RingQueue<Task>;//创建消费者线程for (int i = 0; i < CONSUMRT_NUM; i++){pthread_create(consumer + i, nullptr, ConsumerRoutine, (void *)ring_queue);}//创建生产者线程for (int i = 0; i < PRODUCTER_NUM; i++){pthread_create(producter + i, nullptr, ProducterRoutine, (void *)ring_queue);}//等待消费者线程for (int i = 0; i < CONSUMRT_NUM; i++){pthread_join(consumer[i], nullptr);}//等待生产者线程for (int i = 0; i < PRODUCTER_NUM; i++){pthread_join(producter[i], nullptr);}return 0;
}
多生产和多消费的意义在哪？
提高线程的并发度。
信号量的本质是一个计数器—>计数器的意义是什么？
以前：申请锁—>判断与访问---->释放锁—>本质我们是不清楚临界资源的情况
现在：信号量要提前预定资源的情况，而在PV变化过程中，我们可以在外部就能知晓临界资源的情况。
计数器的意义：可以不用进入临界区，就可以得知临界资源的情况，甚至可以减少内部资源的判断。
