目录
一、POSIX信号量
1. 什么是信号量
在之前的博客中，我们利用加锁解锁保证了每次只有一个线程进入临界资源，但是临界资源很多也很大，如果每次只允许一个线程进入临界资源往往会使效率很低。但是将临界资源划分为多个独立的区域，划分为多少个区域就可以让多少个线程进入。信号量可以理解为一个计数器，它是用来描述临界资源的有效个数；
2. 信号量的基本原理
但是这样就同时带来了一个问题 ---> 如果划分为了5个区域，但是同时进入了10个线程该怎么办?所以这一点可以通过信号量解决。但如何保证信号量是原子性的呢？
P操作：我们将申请信号量称为P操作，申请信号量的本质就是申请获得临界资源中某块资源的使用权限，当申请成功时临界资源中资源的数目应该减一，因此P操作的本质就是让计数器减一。
V操作：我们将释放信号量称为V操作，释放信号量的本质就是归还临界资源中某块资源的使用权限，当释放成功时临界资源中资源的数目就应该加一，因此V操作的本质就是让计数器加一。
- 结合上图和PV操作的理解，我们可以看出，当多个执行流申请信号量时，信号量本质上就是临界资源，对信号量的PV操作表面看似是++和--操作，但是我们知道++和--不是原子性操作，所有我们就要保证PV操作是原子性操作，结合图中右侧的伪代码，可以看出都对PV操作进行的加锁和解锁的操作，这样的目的是为了保证申请和释放信号量时是原子性；
- 当执行流申请信号量时，可能此时信号量为0，说明信号量描述的临界资源被申请完了，那么这个执行流就要挂起等待，在信号量等待队列中等待，直到有信号量释放被唤醒。
- 信号量的本质是计数器，但不意味着只有计数器，信号量还包括一个等待队列。
二、与信号量相关的操作
要使用信号量就需要创建一个 sem_t 类型的变量
#include <semaphore.h>//头文件
sem_t sem1;1. 初始化信号量
在使用信号量前，需要对这个变量进行初始化，使用的函数是 sem_init
int sem_init(sem_t *sem, int pshared, unsigned int value);参数说明：
- sem：需要初始化的信号量。
- pshared：一般给0，传入0值表示线程间共享，传入非零值表示进程间共享。
- value：信号量的初始值（计数器的初始值）。
返回值说明：
- 初始化信号量成功返回0，失败返回-1。
2. 销毁信号量
信号量使用完毕需要用 sem_destory 进行销毁
int sem_destroy(sem_t *sem);参数说明：
- sem：需要销毁的信号量。
返回值说明：
- 销毁信号量成功返回0，失败返回-1。
3. 等待信号量
POSIX信号量中的P操作对应的接口是 sem_waitm
int sem_wait(sem_t* sem);信号量做减1操作
4. 发布信号量
POSIX信号量中的V操作对应的接口是 sem_post
int sem_post(sem_t* sem);信号量做加1操作
三、基于环形队列的生产者消费者模型
基于阻塞队列的生产者与消费者模型存在一个很大的问题就是他们其实是在串行运行的，并没有并行运行，这就导致他们的效率不是很高，而使用环形队列则可以解决这个问题。
这样的模型为什么可以实现并行操作呢?举例来说，当消费者和生产者启动时，由于队列中全部为空，所以即便消费者先运行它也会因为没有数据而被挂起，所以生产者就会先运行生产数据。一旦产生了数据，数据的信号量增加，于是消费者拿到信号进行消费，一旦所有空格都存放了数据，那么生产者就会挂起，当消费者消费完一个数据，然后归还空格，于是生产者又会拿到信号启动生产。这样，只要队列中同时有空格和数据，生产者和消费者就能同时运行。
1. 空间资源和数据资源
生产者关注的是空间资源，消费者关注的是数据资源
- 只要环形队列中有空间，生产者就可以进行生产
- 只要环形队列中有数据，消费者就可以消费数据
2. 生产者和消费者申请和释放资源
我们假设空间资源为 block_sem , 数据资源为 data_sem。在对这两个信号量进行初始化的时候，我们将 block_sem 初始化为环形队列的容量，将 data_sem 初始化为0；（假设环形队列中没有任何数据）
1. 生产者申请空间资源，释放数据资源：
- 如果block_sem不为0，表明环形队列中有空间资源，生产者申请block_sem成功，那么对应的操作就是P(block_sem),向空间内加入数据；然后释放数据资源，即V(data_sem)，此时队列中多了1块空间，那么data_sem就要 –1；
- 如果block_sem为0，那么生产者申请信号量失败，此时生产者就要挂起等待，等待有新的空间资源
2.消费者申请数据资源，释放空间资源：
- 如果data_sem不为0，表明环形队列中有数据，消费者申请data_sem成功，对应的操作时P(data_sem)，从环形队列中取出数据；然后释放空间资源，即V(block_sem)，此时空间资源就多了一个，那么block_sem既要 +1；
- 如果data_sem为0,消费者申请data_sem失败，此时消费者挂起等待，等待新的数据资源。
注意点：
- 如果生产者生产的快，消费者消费的慢，当生产者在生成的过程中遇到了消费者并超过了消费者，那么再生产的数据就会覆盖掉，是绝对不允许的，此时生产者就要挂起等待。
- 如果消费者消费的快，生产者生产的慢，当消费者在消费的过程中遇到了生成者并超过了生产者，那么再消费的数据就有可能是缓存中的废弃数据，是绝对不允许的，此时消费者就要挂起等待。
归根结底，还是环形队列判空判满的一个问题：
上图中，虽然肉眼可见左为空，右为满，但程序不一定能判断出来；所以生产者在生成的时候和消费者在消费的时候，我们要对其下标进行一个合理的控制，确保生产者和消费者之间不会存在冲突。环形队列在之前的博客中有讲到过，这里简单的提一下，判空：生产者和消费指向同一个位置；判满：生产者和消费者之间要预留一个空间；具体操作就是模运算；我们可以看看下面的模拟实现。
四、模拟实现基于环形队列的生产者消费者模型
1. 单生产者与单消费者
我们采用SLT中的vector来实现环形队列
ring_queue.hpp如下：
#pragma once
#include <iostream>
#include <vector>
#include <semaphore.h>
using namespace std;
namespace ns_ring_queue
{
const int g_cap_default = 10; //假设环形队列能存放10个数据
template <class T>
class RingQueue
{
private:
vector<T> ring_queue_; //环形队列
int cap_; //环形队列的容量上限
sem_t blank_sem_; //生产者关心空位置资源（信号量）
sem_t data_sem_; //消费者关心数据（信号量）
int c_step_; //记录消费者的下标
int p_step_; //记录生产者的下标
public:
RingQueue(int cap = g_cap_default):ring_queue_(cap), cap_(cap)
{
sem_init(&blank_sem_, 0, cap); //初始化空间资源（信号量）
sem_init(&data_sem_, 0, 0); //初始化数据资源（信号量）
c_step_ = p_step_ = 0;
}
~RingQueue()
{
sem_destroy(&blank_sem_);
sem_destroy(&data_sem_);
}
public:
void Push(const T& in)//生产接口
{
sem_wait(&blank_sem_);//p空位置（申请空间信号量）
ring_queue_[p_step_] = in; //向环形队列中放数据
sem_post(&data_sem_);//v数据 （发布数据信号量）
p_step_++;
p_step_ %= cap_;
}
void Pop(T* out)//消费接口
{
sem_wait(&data_sem_);//p数据 （申请数据信号量）
*out = ring_queue_[c_step_]; //从环形队列中取数据
sem_post(&blank_sem_);//v空位置 （发布空间信号量）
c_step_++;
c_step_ %= cap_;
}
};
}ring_cp.cc如下：
#include "ring_queue.hpp"
#include <pthread.h>
#include <time.h>
#include <unistd.h>
using namespace ns_ring_queue;
void* consumer(void* args)
{
RingQueue<int>* rq = (RingQueue<int>*)args;
while(true)
{
sleep(1);
int data = 0;
rq->Pop(&data);
printf("消费的数据是：%d\n", data);
}
}
void* producter(void* args)
{
RingQueue<int>* rq = (RingQueue<int>*)args;
while(true)
{
sleep(1);
int data = rand() % 20 + 1;
printf("生产的数据是：%d\n", data);
rq->Push(data);
}
}
int main()
{
srand((long long)time(nullptr));
RingQueue<int>* rq = new RingQueue<int>();
pthread_t c, p;
pthread_create(&c, nullptr, consumer,(void*)rq);
pthread_create(&p, nullptr, producter,(void*)rq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
} 我们将生产者和消费者全部先休眠1秒后再生产和消费数据，运行发现，生产者生产一个数据，消费者消费一个数据
2. 多生产者与多消费者
我们还是以上面的代码为例，我们不在进行单纯的放数据和拿数据，我们让生产者生产出一批计算任务然后让消费者去计算
ring_queue.hpp如下：
#pragma once
#include <iostream>
#include <vector>
#include <semaphore.h>
#include <pthread.h>
using namespace std;
namespace ns_ring_queue
{
const int g_cap_default = 10;
template <class T>
class RingQueue
{
private:
vector<T> ring_queue_;
int cap_;
//生产者关心空位置资源
sem_t blank_sem_;
//消费者关心数据
sem_t data_sem_;
int c_step_;
int p_step_;
pthread_mutex_t c_mtx_;
pthread_mutex_t p_mtx_;
public:
RingQueue(int cap = g_cap_default):ring_queue_(cap), cap_(cap)
{
sem_init(&blank_sem_, 0, cap);
sem_init(&data_sem_, 0, 0);
c_step_ = p_step_ = 0;
pthread_mutex_init(&c_mtx_, nullptr);
pthread_mutex_init(&p_mtx_, nullptr);
}
~RingQueue()
{
sem_destroy(&blank_sem_);
sem_destroy(&data_sem_);
pthread_mutex_destroy(&c_mtx_);
pthread_mutex_destroy(&p_mtx_);
}
public:
void Push(const T& in)
{
//生产接口
sem_wait(&blank_sem_);//p空位置
pthread_mutex_lock(&p_mtx_);
ring_queue_[p_step_] = in;
p_step_++;
p_step_ %= cap_;
pthread_mutex_unlock(&p_mtx_);
sem_post(&data_sem_);//v数据
}
void Pop(T* out)
{
//消费接口
sem_wait(&data_sem_);//p数据
pthread_mutex_lock(&c_mtx_);
*out = ring_queue_[c_step_];
c_step_++;
c_step_ %= cap_;
pthread_mutex_unlock(&c_mtx_);
sem_post(&blank_sem_);//v空位置
}
};
}对于多生产者和多消费者来说，我们要保证他们各自之间要满足互斥，就必须加锁，那么这把锁是在信号量之前加还是之后加呢？
首先明确一点，信号量也是原子性的（上文已经提到过了）
1.在获取信号量之前进行加锁
在这种情况下，也就意味着只有一个执行流能够竞争到锁，然后申请信号量，那么我们对这个临界资源进行划分的意义何在呢，和之前的单生产者单消费者没有太大的区别，显然没有太大的价值；
2.在获取信号量之后进行加锁
在这种情况下，当多个执行流访问临界资源的时候，他们都要去申请信号量，但是只会有一个执行流竞争锁成功，等到这个执行流执行完毕后，下一个执行流就不需要再去申请信号量然后竞争锁，因为它是拿着信号量被挂起的。
总的来说，在获取信号量之后进行加锁，确保了每个执行流都能预定到相应的部分临界资源，相比第一种做法效率高一些；
ring_cp.cc如下：
#include "ring_queue.hpp"
#include "Task.hpp"
#include <time.h>
#include <string>
#include <unistd.h>
using namespace ns_ring_queue;
using namespace ns_task;
void* consumer(void* args)
{
RingQueue<Task>* rq = (RingQueue<Task>*)args;
while(true)
{
sleep(1);
Task t;
rq->Pop(&t);
t();
}
}
void* producter(void* args)
{
RingQueue<Task>* rq = (RingQueue<Task>*)args;
const string ops = "+-*/%";
while(true)
{
sleep(1);
int x = rand() % 20 + 1;
int y = rand() % 10 + 1;
char op = ops[rand() % ops.size()];
Task t(x, y, op);
printf("我生产的数据是：%d %c %d =? 我是：%lu\n",x ,op ,y, pthread_self());
rq->Push(t);
}
}
int main()
{
srand((long long)time(nullptr));
RingQueue<Task>* rq = new RingQueue<Task>();
pthread_t c0, c1, c2, c3, p0, p1, p2;
pthread_create(&c0, nullptr, consumer,(void*)rq);
pthread_create(&c1, nullptr, consumer,(void*)rq);
pthread_create(&c2, nullptr, consumer,(void*)rq);
pthread_create(&c3, nullptr, consumer,(void*)rq);
pthread_create(&p0, nullptr, producter,(void*)rq);
pthread_create(&p1, nullptr, producter,(void*)rq);
pthread_create(&p2, nullptr, producter,(void*)rq);
pthread_join(c0, nullptr);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(p0, nullptr);
pthread_join(p1, nullptr);
pthread_join(p2, nullptr);
return 0;
}Task.hpp如下：
#pragma once
#include <iostream>
#include <pthread.h>
using namespace std;
namespace ns_task
{
class Task
{
private:
int x_;
int y_;
char op_;//用来表示：+ 、- 、* 、/ 、%
public:
Task(){}
Task(int x, int y, char op):x_(x), y_(y), op_(op){}
string show()
{
string message = to_string(x_);
message += op_;
message += to_string(y_);
message += "=?";
return message;
}
int Run()
{
int res = 0;
switch(op_)
{
case '+':
res = x_ + y_;
break;
case '-':
res = x_ - y_;
break;
case '*':
res = x_ * y_;
break;
case '/':
res = x_ / y_;
break;
case '%':
res = x_ % y_;
break;
default:
cout << "bug" << endl;
break;
}
printf("当前任务正在被：%lu处理，处理结果为：%d %c %d = %d\n",pthread_self(), x_, op_, y_, res);
return res;
}
int operator()()
{
return Run();
}
~Task(){}
};
} 运行结果如下：
