目录

信号量的基本操作

基于信号量和环形队列的生产消费模型

---

之前我们在System V版本的进程间通信说过信号量，这里我们说的是POSIX（Portable Operating System Interface for uniX  可移植操作系统接口）版本的并且会说的更加详细

我们之前对于信号量也有一定的理解，我们说它就是要给计数器，用来记录所需资源的数量，因为这个计数器是被多个线程所共享的，所以它本身就是共享资源，为了保证共享资源的安全，所以我们说PV（对计数器--和++）操作是原子的

今天我们还是想用信号量实现生产消费模型，之前我们使用阻塞队列当作生产者和消费者之间的缓冲区，并且这个阻塞队列同时只允许一个线程进入，我们就可以把阻塞队列认为成只有一个位置的信号量，也就是计数器为一

可实际上计数器不一定为一，它可以很大，那么此时我们应该用什么去充当存放任务的容器呢？我们可以选用环形队列，那么我们首先就需要了解一些环形队列的性质，比如我们有p（producer）和c（consumer），它们分别位于环形队列的某一位置，我们可以知道，队列为空和为满时，p、c位于同一位置，也就是说，如果p、c位于不同位置，那么环形队列一定既不为空，也不为满。这时是不是就不会像阻塞队列一样只同时允许一个线程进入了呢，而是一个生产者和一个消费者可以同时进入环形队列。因为不为空不为满时它们一定访问的是不同的位置。这样生产和消费就可以实现真正的并发；如果为空时就让生产者先跑；为满时就让消费者先跑。

也就是说：生产者不能套消费者圈，消费者不能超过生产者

那么我们如何实现上面的东西呢？就是基于信号量

信号量的基本操作

我们需要首先创建一个信号量，就用创建变量的形式去创建就可以

其次要对信号量进行初始化

man sem_init

第二个参数给0，第三个参数是想让计数器是几就填几

申请出信号量要对信号量进行PV操作

man sem_wait

man sem_post

最后要销毁信号量

man sem_destroy

以上就是关于信号量的一些操作，下面我们就实现基于信号量和环形队列的生产消费模型

基于信号量和环形队列的生产消费模型

//Main.cc

#include"RingQueue.hpp"
#include"Task.hpp"
#include"MyThread.hpp"
using namespace MyThread;
#include<vector>
#include<unistd.h>
using namespace std;
void Producercode(RingQueue<Task_t>*prq,string name)
{
    while(1)
    {
        sleep(1);
        prq->Enqueue(print);
        cout<<name<<" produce a task"<<endl;
    }

}
void Consumercode(RingQueue<Task_t>*prq,string name)
{
    sleep(10);
    while(1)
    {
        sleep(1);
        Task_t t;
        prq->Pop(&t);
        cout<<name<<" get a task...";
        t();
    }
}
void InitProducer(vector<Thread<RingQueue<Task_t>*>>&threads,int num,RingQueue<Task_t>*prq)
{
    for(int i=1;i<=num;i++)
    {
        string name="producer thread-"+to_string(i);
        threads.emplace_back(Producercode,prq,name);
    }

}
void InitConsumer(vector<Thread<RingQueue<Task_t>*>>&threads,int num,RingQueue<Task_t>*prq)
{
    for(int i=1;i<=num;i++)
    {
        string name="consumer thread-"+to_string(i);
        threads.emplace_back(Consumercode,prq,name);
    }
}
void StartAll(vector<Thread<RingQueue<Task_t>*>>&threads)
{
    for(auto&e:threads)
    e.start();
}
void WaitAll(vector<Thread<RingQueue<Task_t>*>>&threads)
{
    for(auto&e:threads)
    e.join();
}
int main()
{
    RingQueue<Task_t> rq(5);
    vector<Thread<RingQueue<Task_t>*>>threads;//要把环形队列指针传给各个线程
    InitProducer(threads,3,&rq);
    InitConsumer(threads,4,&rq);

    StartAll(threads);
    WaitAll(threads);



    return 0;
}

//RingQueue.hpp

#include <iostream>
#include <vector>
#include <semaphore.h>
#include <pthread.h>
template <class T>
class RingQueue
{
public:
    RingQueue(int cap = 10)
        : _ring_queue(cap), _cap(cap), _producer_step(0), _consumer_step(0)
    {
        sem_init(&_room_sem, 0, _cap);
        sem_init(&_data_sem, 0, 0);
        pthread_mutex_init(&_producer_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }
    void Enqueue(const T &in)
    {
        sem_wait(&_room_sem);//对空间进行--操作（P操作）
        pthread_mutex_lock(&_producer_mutex);//加锁为了只允许一个生产者进入
        _ring_queue[_producer_step++]=in;
        _producer_step%=_cap;
        pthread_mutex_unlock(&_producer_mutex);
        sem_post(&_data_sem);//对数据进行++操作（V操作）
    }
    void Pop(T *out)
    {
        sem_wait(&_data_sem);
        pthread_mutex_lock(&_consumer_mutex);
        *out=_ring_queue[_consumer_step++];
        _consumer_step%=_cap;
        pthread_mutex_unlock(&_consumer_mutex);
        sem_post(&_room_sem);
    }
    ~RingQueue()
    {
        sem_destroy(&_room_sem);
        sem_destroy(&_data_sem);
        pthread_mutex_destroy(&_producer_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }

private:
    std::vector<T> _ring_queue;
    int _cap;
    // 生产者和消费者的下标
    int _producer_step;
    int _consumer_step;

    sem_t _room_sem; // 生产者关心
    sem_t _data_sem; // 消费者关心

    pthread_mutex_t _producer_mutex; // 生产者之间去竞争锁，竞争锁成功进入环形队列
    pthread_mutex_t _consumer_mutex;
};

//MyThread.hpp

#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>
using namespace std;

namespace MyThread
{
    template<class T>
    using fun_t = function<void(T,string)>;

    template <class T>
    class Thread
    {
    private:
        void excute()
        {
            _func(_pdata,_name);
        }

    public:
        Thread(fun_t<T> func, T pdata, const string&name="noname")
            : _func(func), _pdata(pdata), _name(name), _stop(true) {}

        static void *threadrun(void *args)//如果不是静态，会有this指针
        {
            Thread<T> *ptr = reinterpret_cast<Thread<T> *>(args);
            ptr->excute();
            return nullptr;
        }

        bool start()
        {
            int n = pthread_create(&_id, nullptr, threadrun, this);//把this当参数传过去
            if (n == 0)
            {
                _stop = false;
                return true;
            }
            else
            {
                return false;
            }
        }
        void join()
        {
            if (!_stop)
            {
                pthread_join(_id, nullptr);
            }
        }
        void detach()
        {
            if (!_stop)
            {
                pthread_detach(_id);
            }
        }
        void stop()
        {
            _stop = true;
        }

    private:
        pthread_t _id;
        string _name;
        bool _stop;
        fun_t<T> _func;
        T _pdata;
    };
}

//makefile

cp:Main.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f cp
//task.hpp

#include<iostream>
#include<functional>
using std::cout;
using std::endl;
using Task_t=std::function<void()>;
void print()
{
    cout<<"I am a task..."<<endl;
}