只要我们对资源进行整体加锁就默认了我们对这个资源整体使用，实际情况可能存在一份公共资源，但是允许同时访问不同的区域！(程序员编码保证不同的线程可以并发访问公共资源的不同区域！)
信号量本质是一把计数器，衡量临界资源中资源数量多少的计数器
只要拥有信号量，就在未来一定能够拥有临界资源的一部分，申请信号量的本质：对临界资源中特定小块资源的预定机制。比如电影院买票预定座位
只要申请成功，就一定有你的资源，只要申请失败，就说明条件不就绪，你只能等，就不需要判断了
线程要进行访问临界资源中的某一区域——得先申请信号量——前提是所有人必须先看到信号量——所以信号量本身必须是：公共资源。
信号量的核心操作就是PV原语
PV操作必须是原子的：
多个执行流为了访问临界资源会竞争式的申请信号量，因此信号量是会被多个执行流同时访问的，也就是说信号量本质也是临界资源。
但信号量本质就是用于保护临界资源的，我们不可能再用信号量去保护信号量，所以信号量的PV操作必须是原子操作。
注意： 内存当中变量的++、–操作并不是原子操作，因此信号量不可能只是简单的对一个全局变量进行++、–操作。
sem_init 初始化信号量：
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数说明：
sem：需要初始化的信号量。
pshared：传入0值表示线程间共享，传入非零值表示进程间共享。
value：信号量的初始值（计数器的初始值）。
返回值说明：
初始化信号量成功返回0，失败返回-1。
注意： POSIX信号量和System V信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的，但POSIX信号量可以用于线程间同步。
sem_destroy 销毁信号量：
int sem_destroy(sem_t *sem);
参数说明：
sem：需要销毁的信号量。
返回值说明：
销毁信号量成功返回0，失败返回-1。
sem_wait 等待信号量（申请信号量）：
int sem_wait(sem_t *sem);
参数说明：
sem：需要等待的信号量。
返回值说明：
等待信号量成功返回0，信号量的值减一。
等待信号量失败返回-1，信号量的值保持不变。
sem_post 发布信号量（释放信号量）：
int sem_post(sem_t *sem);
参数说明：
sem：需要发布的信号量。
返回值说明：
发布信号量成功返回0，信号量的值加++。
发布信号量失败返回-1，信号量的值保持不变。
上一篇博文的生产者-消费者的例子是基于queue的,其空间可以动态分配,现在基于固定大小的环形队列重写这个程序。环形队列采用数组模拟（推荐数组模拟，因为缓存命中率高），用模运算来模拟环状特性。此环形队列的生产者消费者模型中，生产线程用来放数据，消费线程用来拿数据。
引入环形队列
环形队列之前我们就了解过了，只要是环形队列，就存在判空判满的问题。实际上并不是真正的环形队列，而是通过数组模拟的，当数据加入到最后的位置时直接模等于数组的大小即可。通常情况下，判空判满的问题我们是通过空出一个位置，当两个指针指向同一个位置的时候是空，当只剩一个位置的时候就是满，但是我们这里不需要关注。
访问环形队列
生产者和消费者访问同一个位置的情况：空的时候，满的时候；其他情况下生产者与消费者访问的就是不同的区域了。
为了完成环形队列的生产消费，我们的核心工作就是
1.消费者不能超过生产者
2.生产者不能套消费者一个圈以上
3.生产者和消费者指向同一个位置时，如果此时满了就让消费者先走，如果此时为空就让生产者先走
我们现在不需要考虑环形队列为满为空的，因为有信号量帮我们考虑。当我们不考虑留一个空位置的时候，发生两个执行流访问同一个位置只有当环形队列为满或为空的情况（互斥 & 同步）。其它的时候，我们都指向不同的位置。
当空的时候，消费者不能消费，而要让生产者生产，当满的时候，生产者不能生产，而要让消费者消费，这是互斥的体现
我们不能让二者同时访问一个位置并执行，要具有一定的顺序写，这是同步的体现
当二者指向不同的位置时，二者可以同时运行，这是并发的体现。
如果生产者和消费者两个执行流指向了同一个位置，那么它们谁先运行呢？解决此问题，需要分类讨论（为空 & 为满），答案如下：
注意：上述的原则是由信号量来保证的。信号量是用来描述临界资源数目多少的计数器。我们通过信号量保证了生产者和消费者的串行化过程。而除了这两种情况之外，生产者和消费者指向的都不是同一个位置，因此该环形队列当中不可能会出现数据不一致的问题。并且大部分情况下生产者和消费者指向并不是同一个位置，因此大部分情况下该环形队列可以让生产者和消费者并发的执行
大部分情况下生产者与消费者是并发执行的，但是当环形队列为空或为满的时候就会存在着同步与互斥问题。
如何去进行保证：信号量维护，信号量是衡量临界资源中资源数量的
资源是什么：
1.对于生产者，看中的是队列中的剩余空间，空间资源定义成一个信号量
2.对于消费者，看中的是队列中的数据资源，数据资源定义成一个信号量
比如我们一共有10个位置，消费者初始信号量是0，生产者初始信号量是10，如果生产者线程生产数据，申请信号量，进行P操作，信号量变为9，申请失败则阻塞；申请成功后消费者线程看到了多一个数据资源，消费者信号量进行V操作.所以我们并不需要进行判空判满：当生产者生产满了，信号量申请不到，进行阻塞，只能让消费者先走；当消费者消费完了，信号量申请不到，只能让生产者先走
生产者和消费者的申请和释放资源
对于生产者和消费者来说，它们关注的资源是不同的：（我们假设环形队列的空间是N）
生产者关注的是环形队列当中是否有空间（room），只要有空间生产者就可以进行生产。其空间变化应该是 N -> 0
消费者关注的是环形队列当中是否有数据（data），只要有数据消费者就可以进行消费。其数据变化是0 -> N
现在我们用信号量来描述环形队列当中的空间资源（roomSem）和数据资源（dataSem），在我们初始信号量时给它们设置的初始值是不同的：
roomSem的初始值我们应该设置为环形队列的容量，因为刚开始时唤醒队列中全是空间
dataSem的初始值我们应该设置为0，因为刚开始时环形队列中没有数据
生产者申请空间资源，释放数据资源：
对于生产者而言，每次生产数据前要先申请空间资源rootSem
对于消费者而言，每次消费前要先申请数据资源dataSem
如果dataSem的值不为0，有数据，可以消费数据，则信号量申请成功，此时消费者可以进行消费操作。
如果dataSem的值为0，没有数据，无法消费数据，则信号量申请失败，此时消费者需要在dataSem的等待队列下进行阻塞等待，直到环形队列当中有新的数据后再被唤醒（当生产者生产数据后）。
当消费者消费完数据后，应该释放rootSem：
虽然消费者在进行消费前是对dataSem进行的P操作，但是当消费者消费完数据，应该对rootSem进行V操作而不是dataSem。因为消费完数据后，空间是会多一个，所以对rootSem进行V操作
消费者在消费数据前申请到的是data位置，当消费者消费完数据后，该位置当中的数据已经被消费过了，再次被消费就没有意义了，为了让生产者后续可以在该位置生产新的数据，我们应该将该位置算作room位置，而不是data位置。
当消费者消费完数据后，意味着环形队列当中多了一个room位置，因此我们应该对rootSem进行V操作。
RingQueue.hpp文件：
上述的环形队列RingQueue就是生产者消费者模型当中的交易场所，我们可以用C++SEL库当中的vector进行实现。我们将RingQueue设计出模板类。
RingQueue的私有成员变量如下：
ringqueue_：此环形队列用数组实现的，所以我们定义一个vector数组ringqueue_
roomSem_ & dataSem_：分别用于生产者衡量空间计数器的信号量和消费者衡量数据计数器的信号量
pIndex_ & cIndex_：即下标，分别记录当前生产者写入的位置和消费者读取的位置
RingQueue的公有成员函数如下：
RingQueue构造函数：
给数组ringqueue_初始化容量为5（定义全局变量gCap），初始化消费者生产者下标为0
复用sem_init对roomSem和dataSem进行初始化
~RingQueue析构函数：
复用sem_destroy释放空间信号量和数据信号量
push生产函数：
复用sem_wait函数申请roomSem空间资源
接着把生产出的数据放入ringqueue_数组里，更新pIndex_++，为了防止pIndex_后续越界，要 % 数组ringqueue_的大小size()
创建数据后，复用sem_post函数对dataSem信号量进行V操作，因为生产数据后，数据信号量要++
pop消费函数：
复用sem_wait函数申请dataSem数据资源
定义临时变量保存要拿出的数据，拿出后不需要释放此数据，因为往后生产者生产数据会覆盖此位置
拿走数据后，空间就多了一个，复用sem_post函数对roomSem信号量进行V操作
拿走数据后，更新cIndex的位置，为了防止cIndex_后续越界，要 % 数组ringqueue_的大小size()
返回temp拿走的数据
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
using namespace std;
// 定义默认容量大小
const int gCap = 5;
template <class T>
class RingQueue
{
public:
// 构造函数
RingQueue(int cap = gCap)
: ringqueue_(cap), pIndex_(0), cIndex_(0)
{
// 初始化空间计数器信号量
sem_init(&roomSem_, 0, ringqueue_.size());
// 初始化数据计数器信号量
sem_init(&dataSem_, 0, 0);
}
// 析构函数
~RingQueue()
{
sem_destroy(&roomSem_);
sem_destroy(&dataSem_);
}
// 生产
void push(const T &in)
{
sem_wait(&roomSem_); // 申请空间资源信号量
ringqueue_[pIndex_] = in; // 生产数据放入数组里
sem_post(&dataSem_); // V操作
pIndex_++; // 更新写入的位置
pIndex_ %= ringqueue_.size(); // 确保pIndex_的有效性
}
// 消费
T pop()
{
sem_wait(&dataSem_); // 申请数据资源信号量
T temp = ringqueue_[cIndex_]; // 保存拿的数据
sem_post(&roomSem_); // V操作
cIndex_++; // 更新读取的位置
cIndex_ %= ringqueue_.size(); // 确保cIndex_的有效性
return temp;
}
private:
vector<T> ringqueue_; // 环形队列
sem_t roomSem_; // 衡量空间计数器, productor
sem_t dataSem_; // 衡量数据计数器, consumer
uint32_t pIndex_; // 当前生产者写入的位置
uint32_t cIndex_; // 当前消费者读取的位置
};
RingQueueTest.cc文件：
环形队列需要让生产者不断生产数据，消费者不断消费数据。我们利用rand函数生成一个随机数，并让生产者把此数据push到环形队列里，而消费者就是不断从环形队列里pop数据。为了便于观察，我们可以将生产者生产的数据和消费者消费的数据进行打印输出。
#include "RingQueue.hpp"
#include <unistd.h>
#include <ctime>
// 生产者不断生产数据
void *productor(void *args)
{
RingQueue<int> *rqp = static_cast<RingQueue<int> *>(args);
while (true)
{
int data = rand() % 10;
rqp->push(data);
cout << "pthread[" << pthread_self() << "]" << " 生产了一个数据: " << data << endl;
sleep(1);
}
}
// 消费者不断消费数据
void *consumer(void *args)
{
RingQueue<int> *rqp = static_cast<RingQueue<int> *>(args);
while (true)
{
int data = rqp->pop();
cout << "pthread[" << pthread_self() << "]" << " 消费了一个数据: " << data << endl;
}
}
int main()
{
srand((unsigned long)time(nullptr) ^ getpid());
RingQueue<int> rq;
pthread_t c, p;
pthread_create(&p, nullptr, productor, &rq);
pthread_create(&c, nullptr, consumer, &rq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}
上述代码我们实现的是单生产者单消费者，现在来实现一个多生产者多消费者。对于多生产者和多消费者，要遵循生产者与生产者之间互斥、消费者与消费者之间互斥、生产者与消费者之间互+同步的原则。逻辑变化如下：
加锁：
对于push生产函数和pop消费函数，已进入函数都要申请信号量资源，而多生产者多消费者就会导致多个线程同时申请信号资源进行P操作，P操作是原子的，随后就是多个线程在pIndex_位置放数据和cIndex_位置拿数据，此时的pIndex_和cIndex_就变成了临界资源，后续在访问临界区的时候，要维护生产者和生产者之间，消费者和消费者之间的互斥性。最多只允许一个线程对此临界区的写入操作。所以要对此临界区加锁。
push和pop加锁的位置：
申请信号量是资源的预定机制，因此所有线程都可以先申请信号量。最好把加锁放到申请信号量后面，至于哪个线程申请到锁，这是它们内部竞争决定的。这样做就好比如它们都拿到了门票（申请信号量），最后一个个进来（申请锁，访问临界区），此加锁过程对于pop函数也亦是如此。
RingQueue.hpp文件：
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
using namespace std;
// 定义默认容量大小
const int gCap = 5;
template <class T>
class RingQueue
{
public:
// 构造函数
RingQueue(int cap = gCap)
: ringqueue_(cap), pIndex_(0), cIndex_(0)
{
// 初始化空间资源和数据资源信号量
sem_init(&roomSem_, 0, ringqueue_.size());
sem_init(&dataSem_, 0, 0);
// 初始化生产者和消费者的锁
pthread_mutex_init(&pmutex_, nullptr);
pthread_mutex_init(&cmutex_, nullptr);
}
// 析构函数
~RingQueue()
{
//释放信号量
sem_destroy(&roomSem_);
sem_destroy(&dataSem_);
//释放锁
pthread_mutex_destroy(&pmutex_);
pthread_mutex_destroy(&cmutex_);
}
// 生产
void push(const T &in)
{
sem_wait(&roomSem_); // 申请空间资源信号量
pthread_mutex_lock(&pmutex_);
ringqueue_[pIndex_] = in; // 生产数据放入数组里
pIndex_++; // 更新写入的位置
pIndex_ %= ringqueue_.size(); // 确保pIndex_的有效性
pthread_mutex_unlock(&pmutex_);
sem_post(&dataSem_); // V操作
}
// 消费
T pop()
{
sem_wait(&dataSem_); // 申请数据资源信号量
pthread_mutex_lock(&cmutex_);
T temp = ringqueue_[cIndex_]; // 保存拿的数据
cIndex_++; // 更新读取的位置
cIndex_ %= ringqueue_.size(); // 确保cIndex_的有效性
pthread_mutex_unlock(&cmutex_);
sem_post(&roomSem_); // V操作
return temp;
}
private:
vector<T> ringqueue_; // 环形队列
sem_t roomSem_; // 衡量空间计数器, productor
sem_t dataSem_; // 衡量数据计数器, consumer
uint32_t pIndex_; // 当前生产者写入的位置, 如果是多线程, pIndex_也是临界资源
uint32_t cIndex_; // 当前消费者读取的位置, 如果是多线程, cIndex_也是临界资源
pthread_mutex_t pmutex_; // 生产者的锁
pthread_mutex_t cmutex_; // 消费者的锁
};
RingQueueTest.cc文件：
#include "RingQueue.hpp"
#include <unistd.h>
#include <ctime>
// 生产者不断生产数据
void *productor(void *args)
{
RingQueue<int> *rqp = static_cast<RingQueue<int> *>(args);
while (true)
{
int data = rand() % 10;
rqp->push(data);
cout << "pthread[" << pthread_self() << "]" << " 生产了一个数据: " << data << endl;
sleep(1);
}
}
// 消费者不断消费数据
void *consumer(void *args)
{
RingQueue<int> *rqp = static_cast<RingQueue<int> *>(args);
while (true)
{
int data = rqp->pop();
cout << "pthread[" << pthread_self() << "]" << " 消费了一个数据: " << data << endl;
}
}
int main()
{
srand((unsigned long)time(nullptr) ^ getpid());
RingQueue<int> rq;
pthread_t c1, c2, c3, p1, p2, p3;
pthread_create(&p1, nullptr, productor, &rq);
pthread_create(&p2, nullptr, productor, &rq);
pthread_create(&p3, nullptr, productor, &rq);
pthread_create(&c1, nullptr, consumer, &rq);
pthread_create(&c2, nullptr, consumer, &rq);
pthread_create(&c3, nullptr, consumer, &rq);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(p1, nullptr);
pthread_join(p2, nullptr);
pthread_join(p3, nullptr);
return 0;
}
tr, productor, &rq);
pthread_create(&p3, nullptr, productor, &rq);
pthread_create(&c1, nullptr, consumer, &rq);
pthread_create(&c2, nullptr, consumer, &rq);
pthread_create(&c3, nullptr, consumer, &rq);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(p1, nullptr);
pthread_join(p2, nullptr);
pthread_join(p3, nullptr);
return 0;
}
