POSIX信号量

初始化信号量

include

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);

‍

基于环形队列的⽣产消费模型

Mutex.hpp

#pragma once

#include<iostream>
#include<pthread.h>

class Mutex
{
public:
    Mutex()
    {
        pthread_mutex_init(&_lock , nullptr);
    }

    void Lock()
    {
        pthread_mutex_lock(&_lock);
    }

    pthread_mutex_t *Ptr()
    {
        return &_lock;
    }

    void Unlock()
    {
        pthread_mutex_unlock(&_lock);
    }

    ~Mutex()
    {
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

class LockGuard
{
public:
    LockGuard(Mutex &lock):_lockref(lock)
    {
        _lockref.Lock();
    }

    ~LockGuard()
    {
        _lockref.Unlock();
    }
private:
    Mutex &_lockref;
};

‍

Sem.hpp

#ifndef __SEM_HPP
#define __SEM_HPP

#include<iostream>
#include<semaphore.h>

class Sem
{
public:
    Sem(int init_val)
    {
        if(init_val > 0)
        {
            int n = sem_init(&_sem , 0 , init_val);
            (void)n;
        }
    }

    void P()
    {
        int n = sem_wait(&_sem);
        (void)n;
    }

    void V()
    {
        int n = sem_post(&_sem);
        (void)n;
    }

    ~Sem()
    {
        int n = sem_destroy(&_sem);
        (void)n;
    }
private:
    sem_t _sem;
};

#endif

‍

RingQueue.hpp

#pragma once

#include<iostream>
#include<string>
#include<pthread.h>
#include<vector>
#include"Sem.hpp"
#include"Mutex.hpp"


const int defaultcap = 5;

template<typename T>
class RingQueue
{
public:
    RingQueue(int cap = defaultcap):
    _cap(cap),
    _rq(cap),
    _consumer_step(0),
    _productor_step(0),
    _blank_sem(cap),
    _data_sem(0)
    {}

    void Enqueue(T &in)
    {
        // 1. 预定资源
        _blank_sem.P();

        {
            LockGuard lockguard(_pmutex);
            // 2. 定位生成位置
            _rq[_productor_step++] = in;
            _productor_step %= _cap;

        }

        // 3. 释放数据资源
        _data_sem.V();    
    }

    void Pop(T *out)
    {
        _data_sem.P();

        {
            LockGuard lockguard(_cmutex);
            *out = _rq[_consumer_step++];
            _consumer_step %= _cap;
        }

        _blank_sem.V();
    }

    ~RingQueue()
    {

    }
private:
    int _cap; // 环形队列容量
    std::vector<T> _rq;

    int _consumer_step; //消费位置
    int _productor_step; //生产位置

    Sem _blank_sem; // 格子资源计数器
    Sem _data_sem; //数据信号量

    Mutex _cmutex;
    Mutex _pmutex;

};

Task.hpp

#ifndef __TASK_HPP
#define __TASK_HPP

#include<iostream>
#include<string>
#include<functional>


// using task_t = std::function<void()>;

// void Print()
// {
//     std::cout<<"待处理任务" <<std::endl;
// }

class Task
{
public:
    Task(){}

    Task(int x , int y):_x(x) , _y(y)
    {}

    void Execute()
    {
        _result = _x + _y;
    }

    void operator()()
    {
        Execute();
    }

    std::string getResult()
    {
        return std::to_string(_x) + " + " + std::to_string(_y) + " = " +std::to_string(_result);         
    }

    std::string Question()
    {
        return std::to_string(_x) + " + " + std::to_string(_y) + " =?";
    }

    ~Task()
    {}

private:
    int _x;
    int _y;
    int _result;
};





#endif

‍

Main.cc

#include"RingQueue.hpp"
#include"Task.hpp"
#include <unistd.h>
#include<ctime>
#include<stdlib.h>

// Mutex cnt_lock;
// Mutex screen_lock;

// int data = 1;

// int GetData()
// {
//     cnt_lock.Lock();
//     int result = data++;
//     cnt_lock.Unlock();
//     return result;
// }

// void Print(const std::string name , const std::string &info)
// {
//     screen_lock.Lock();
//     std::cout<<name <<" : " <<info <<std::endl;
//     screen_lock.Unlock();
// }

// class ThreadData
// {
// public:
//     ThreadData(RingQueue<int> *r , const std::string &n):rq(r) , name(n)
//     {
// 
//     }
// 
//     ~ThreadData()
//     {
// 
//     }
// public:
//     std::string name;
//     RingQueue<int> *rq;
// };


void *ProductorRoutine(void *args)
{
    RingQueue<Task> *rq = static_cast<RingQueue<Task> *>(args);

    while(true)
    {
        int x = rand() % 9 + 1;
        usleep(1000);
        int y = rand() % 9 + 1;
        Task t(x , y);

        rq->Enqueue(t);
        std::cout<<"生产任务" <<t.Question() <<std::endl;
    }

    // ThreadData *td = static_cast<ThreadData*>(args);
    // pthread_setname_np(pthread_self() , td->name.c_str());
    // while(true)
    // {
    //     int data = GetData();
    //     sleep(1);
    //     td->rq->Enqueue(data);
    //     Print(td->name , " 生产数据 " + std::to_string(data));
    // }
}

void *ConsumerRoutine(void* args)
{
    RingQueue<Task> *rq = static_cast<RingQueue<Task> *>(args);

    int data = 0;

    while(true)
    {
        sleep(1);

        Task t;
        rq->Pop(&t);

        t();
        std::cout<<"消费完成任务 " <<t.getResult() <<std::endl;
    }

    // ThreadData *td = static_cast<ThreadData*>(args);
    // pthread_setname_np(pthread_self() , td->name.c_str());
    // 
    // int data = 0;
    // while(true)
    // {
    //     // sleep(1);
    //     td->rq->Pop(&data);
    //     Print(td->name , " 消费数据 " + std::to_string(data));
    // }
}

int main()
{
    srand(time(nullptr) ^ getpid());

    RingQueue<int> *rq = new RingQueue<int>();
    pthread_t c , p;
    pthread_create(&p , nullptr , ProductorRoutine , rq);
    pthread_create(&c , nullptr , ConsumerRoutine , rq);


    pthread_join(c , nullptr);
    pthread_join(p , nullptr);

    // ThreadData *td0 = new ThreadData(rq , "product-1");
    // pthread_create(p , nullptr , ProductorRoutine , td0);
    // ThreadData *td1 = new ThreadData(rq , "product-2");
    // pthread_create(p+1 , nullptr , ProductorRoutine , td1);
    // ThreadData *td2 = new ThreadData(rq , "product-3");
    // pthread_create(p+2 , nullptr , ProductorRoutine , td2);

    // ThreadData *td3 = new ThreadData(rq , "consumer-1");
    // pthread_create(c , nullptr , ConsumerRoutine , td3);
    // ThreadData *td4 = new ThreadData(rq , "consumer-2");
    // pthread_create(c+1 , nullptr , ConsumerRoutine , td4);



    // pthread_join(c[0] , nullptr);
    // pthread_join(c[1] , nullptr);
    // pthread_join(p[0] , nullptr);
    // pthread_join(p[1] , nullptr);
    // pthread_join(p[2] , nullptr);

    return 0;
}

‍

makefile

cp_ring:Main.cc
    g++ -o $@ $^ -std=c++14
.PHONY:clean
clean:
    rm -f cp_ring

‍