Linux线程同步与生产消费者模型

#include<iostream>
#include<pthread.h>
#include<unistd.h>
using namespace std;
int count = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//初始化锁
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;//初始化条件变量

void* Count(void* args)
{
    pthread_detach(pthread_self());
    uint64_t number = (uint64_t)args;
    while(1)
    {
        pthread_mutex_lock(&mutex);
        pthread_cond_wait(&cond,&mutex);//让线程等待的时候会自动释放锁，如果不释放锁，lock外面的进不来
        cout << "pthread:" << number << "count:" << count++ << endl;
        pthread_mutex_unlock(&mutex);

    }
}
int main()
{
    for(int i = 0;i < 5;i++)
    {
        pthread_t tid;
        pthread_create(&tid,nullptr,Count,(void*)i);
    }
    while(1) 
    {
        sleep(1);
        pthread_cond_signal(&cond);//主线程唤醒等待队列中的其他线程，默认唤醒第一个
        cout << "signal one pthread" << endl; 
    }
    return 0;
}

//test.hpp
#include<iostream>
#include<pthread.h>
#include<unistd.h>
#include<queue>
using namespace std;
template<class T>
class BlockQueue
{
    static const int defalutnum = 5;
public:
    BlockQueue(int maxcap = defalutnum):maxcap_(maxcap)
    {
        pthread_mutex_init(&mutex_,nullptr);
        pthread_cond_init(&c_,nullptr);
        pthread_cond_init(&p_,nullptr);

    }
    const T pop()//消费
    {
        pthread_mutex_lock(&mutex_);
        if(q_.size() == 0)
        {
            pthread_cond_wait(&c_,&mutex_);//想消费需要满足条件，如果没有资源，就不满足，需要进入队列等待资源补充
        }
        T out  = q_.front();
        q_.pop();
        pthread_cond_signal(&p_);//顺利消费一个说明队列不为满，通知生产者进行生产
        pthread_mutex_unlock(&mutex_);
        return out;
    }
    void push(const T &in)//生产
    {
        pthread_mutex_lock(&mutex_);
        if(q_.size() == maxcap_)//
        {
            pthread_cond_wait(&p_,&mutex_);//想生产需要满足条件，如果资源已满，就不满足，需要进入队列等待资源被拿走
        }
        q_.push(in);
        pthread_cond_signal(&c_);//顺利生产完一个说明队列里面有资源，通知消费者消费
        pthread_mutex_unlock(&mutex_);
    }
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&c_);
        pthread_cond_destroy(&p_);
    }
private:
    queue<T> q_;//共享资源
    int maxcap_;//资源数量极值

    pthread_mutex_t mutex_;
    pthread_cond_t c_;
    pthread_cond_t p_;

};

//main.cc
#include "test.hpp"

void* Consunmer(void* args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);
    while(1)//消费
    {
        int data = bq->pop();
        cout << "消费了一个数据" << data << endl;
    }
}
void* Productor(void* args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);
    int data = 0;
    while(1)//生产
    {
        sleep(1);
        data++;
        bq->push(data);
        cout << "生产了一个数据" << data << endl;
    }
}
int main()
{
    BlockQueue<int> *bq = new BlockQueue<int>();
    pthread_t c,p;
    pthread_create(&c,nullptr,Consunmer,bq);
    pthread_create(&p,nullptr,Productor,bq);

    pthread_join(c,nullptr);
    pthread_join(p,nullptr);
    delete bq;
    return 0;
}

#include<iostream>
#include<pthread.h>
#include<unistd.h>
#include<queue>
#include<semaphore.h>
#include<ctime>
using namespace std;

const static int defaultcap = 5;
template<class T>
class RingQueue
{
private:
    void P(sem_t &sem)
    {
        sem_wait(&sem);
    }
    void V(sem_t &sem)
    {
        sem_post(&sem);
    }
public:
    RingQueue(int cap = defaultcap)
    :ringqueue_(cap),cap_(cap),c_step_(0),p_step_(0)
    {
        sem_init(&cdata_sem,0,0);
        sem_init(&pspace_sem,0,cap);

    }
    void push(const T& val)
    {
        P(pspace_sem);
        ringqueue_[p_step_] = val;
        p_step_++;
        p_step_ %= cap_;
        V(cdata_sem);


    }
    void pop(T *out)
    {
        P(cdata_sem);
        *out = ringqueue_[c_step_];
        c_step_++;
        c_step_ %= cap_;
        V(pspace_sem);


    }
    ~RingQueue()
    {
        sem_destroy(&cdata_sem);
        sem_destroy(&pspace_sem);
    }
private:
    vector<T> ringqueue_;
    int cap_;

    int c_step_;//消费者下标
    int p_step_;//生产者下标

    sem_t cdata_sem;//消费者关注的数据资源
    sem_t pspace_sem;//生产者关注的空间资源
};

#include "RingQueue.hpp"

void* Consunmer(void* args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int>*>(args);
    int data = 0;
    while(1)//消费
    {
        
        rq->pop(&data);
        cout << "Consunmer..." << data << endl;
        sleep(1);
        
    }
}
void* Productor(void* args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int>*>(args);
    int data = 0;
    sleep(3);//这里停顿3S，消费者3S中内也并没有运行
    while(1)//生产
    {
        data = rand() %10;
        rq->push(data);
        cout << "Productor..." << data << endl;
        sleep(1);
    }
}

int main()
{
    srand(time(nullptr)^getpid());
    RingQueue<int>* rq = new RingQueue<int>();
    pthread_t c,p;
    pthread_create(&c,nullptr,Consunmer,rq);
    pthread_create(&p,nullptr,Productor,rq);

    pthread_join(c,nullptr);
    pthread_join(p,nullptr);
    delete rq;
    return 0;
}

template <typename T>
class Singleton {
static T data;
public:
static T* GetInstance() {
return &data;
}
};

template <typename T>
class Singleton {
static T* inst;
public:
static T* GetInstance() {
if (inst == NULL) {
inst = new T();
}
return inst;
}
};

template <typename T>
class Singleton {
volatile static T* inst; // 需要设置 volatile 关键字, 否则可能被编译器优化.
static std::mutex lock;
public:
static T* GetInstance() {
if (inst == NULL) { // 双重判定空指针, 降低锁冲突的概率, 提高性能.
lock.lock(); // 使用互斥锁, 保证多线程情况下也只调用一次 new.
if (inst == NULL) {
inst = new T();
}
lock.unlock();
}
return inst;
}
};