文章目录
- 信号量概念
- 信号量接口
- 初始化
- 销毁
- 等待
- 发布
- 基于环形队列的生产者消费者模型
- 编码
- 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变化过程中，我们可以在外部就能知晓临界资源的情况。
计数器的意义：可以不用进入临界区，就可以得知临界资源的情况，甚至可以减少内部资源的判断。
