【Linux】从多线程同步到生产者消费者模型：多线程编程实践

#ifndef __BLOCK_QUEUE_HPP__
#define __BLOCK_QUEUE_HPP__

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

template <typename T>
class BlockQueue
{
private:
    bool IsFull()
    {
        return _block_queue.size() == _cap;
    }
    bool IsEmpty()
    {
        return _block_queue.empty();
    }
public:
    BlockQueue(int cap) : _cap(cap)
    {
        _productor_wait_num = 0;
        _consumer_wait_num = 0;
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_product_cond, nullptr);
        pthread_cond_init(&_consum_cond, nullptr);
    }
    void Enqueue(T &in) // 生产者用的接口
    {
        pthread_mutex_lock(&_mutex);
        while(IsFull()) // 保证代码的健壮性
        {
            // 生产线程去等待,是在临界区中休眠的！你现在还持有锁呢！！！
            // 1. pthread_cond_wait调用是: a. 让调用线程等待 b. 自动释放曾经持有的_mutex锁 c. 当条件满足，线程唤醒，pthread_cond_wait要求线性
            // 必须重新竞争_mutex锁，竞争成功，方可返回！！！
            // 之前：安全
            _productor_wait_num++;
            pthread_cond_wait(&_product_cond, &_mutex);  //  只要等待，必定会有唤醒，唤醒的时候，就要继续从这个位置向下运行！！
            _productor_wait_num--;
            // 之后：安全
        }
        // 进行生产
        // _block_queue.push(std::move(in));
        // std::cout << in << std::endl;
        _block_queue.push(in);
        // 通知消费者来消费
        if(_consumer_wait_num > 0)
            pthread_cond_signal(&_consum_cond); // pthread_cond_broadcast
        pthread_mutex_unlock(&_mutex);
    }
    void Pop(T *out) // 消费者用的接口 --- 5个消费者
    {
        pthread_mutex_lock(&_mutex);
        while(IsEmpty()) // 保证代码的健壮性
        {
            // 消费线程去等待,是在临界区中休眠的！你现在还持有锁呢！！！
            // 1. pthread_cond_wait调用是: a. 让调用进程等待 b. 自动释放曾经持有的_mutex锁
            _consumer_wait_num++;
            pthread_cond_wait(&_consum_cond, &_mutex);  // 伪唤醒
            _consumer_wait_num--;
        }

        // 进行消费
        *out = _block_queue.front();
        _block_queue.pop();
        // 通知生产者来生产
        if(_productor_wait_num > 0)
            pthread_cond_signal(&_product_cond);
        pthread_mutex_unlock(&_mutex);
        // pthread_cond_signal(&_product_cond);
    }
    ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_product_cond);
        pthread_cond_destroy(&_consum_cond);
    }

private:
    std::queue<T> _block_queue;   // 阻塞队列，是被整体使用的！！！
    int _cap;                     // 总上限
    pthread_mutex_t _mutex;       // 保护_block_queue的锁
    pthread_cond_t _product_cond; // 专门给生产者提供的条件变量
    pthread_cond_t _consum_cond;  // 专门给消费者提供的条件变量

    int _productor_wait_num;
    int _consumer_wait_num;
};

#endif

int sem_wait(sem_t *sem);

int sem_post(sem_t *sem);

#pragma once

#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
#include <pthread.h>


template<typename T>
class RingQueue
{
private:
    void P(sem_t &sem)
    {
        sem_wait(&sem);
    }
    void V(sem_t &sem)
    {
        sem_post(&sem);
    }
    void Lock(pthread_mutex_t &mutex)
    {
        pthread_mutex_lock(&mutex);
    }
    void Unlock(pthread_mutex_t &mutex)
    {
        pthread_mutex_unlock(&mutex);
    }
public:
    RingQueue(int cap): _ring_queue(cap), _cap(cap),  _productor_step(0), _consumer_step(0)
    {
        sem_init(&_room_sem, 0, _cap);
        sem_init(&_data_sem, 0, 0);

        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }
    void Enqueue(const T &in)
    {
        // 生产行为
        P(_room_sem);
        Lock(_productor_mutex);
        // 一定有空间！！！
        _ring_queue[_productor_step++] = in; // 生产
        _productor_step %= _cap;
        Unlock(_productor_mutex);
        V(_data_sem);
    }
    void Pop(T *out)
    {
        // 消费行为
        P(_data_sem);
        Lock(_consumer_mutex);
        *out = _ring_queue[_consumer_step++];
        _consumer_step %= _cap;
        Unlock(_consumer_mutex);
        V(_room_sem);
    }
    ~RingQueue()
    {
        sem_destroy(&_room_sem);
        sem_destroy(&_data_sem);

        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }
private:
    // 1. 环形队列
    std::vector<T> _ring_queue;
    int _cap; // 环形队列的容量上限

    // 2. 生产和消费的下标
    int _productor_step;
    int _consumer_step;

    // 3. 定义信号量
    sem_t _room_sem; // 生产者关心
    sem_t _data_sem; // 消费者关心

    // 4. 定义锁，维护多生产多消费之间的互斥关系
    pthread_mutex_t _productor_mutex;
    pthread_mutex_t _consumer_mutex;
};