基于阻塞队列的生产者-消费者模型实现详解

创作时间:

作者:

@小白创作中心

基于阻塞队列的生产者-消费者模型实现详解

引用

CSDN

https://blog.csdn.net/2301_79465388/article/details/146163027

阻塞队列（Blocking Queue）是实现生产者-消费者模型的重要数据结构，它允许生产者和消费者并发地进行数据生产和消费操作。本文将详细介绍基于阻塞队列的生产者-消费者模型的实现，包括模型的特点、代码实现以及多线程环境下的性能优化。

生产者消费者模型特点

生产者-消费者模型是一种经典的并发设计模式，其主要特点如下：

一个交易场所：以特定数据结构形式存在的一段内存空间
两种角色：生产者（生产线程）和消费者（消费线程）
三种关系：

生产者之间
消费者之间
生产者和消费者之间的关系

与阻塞队列结合后形成的特点：

同步生产和消费操作：在多线程环境下，生产者和消费者可以同时操作队列，但需要保证线程安全。
避免资源浪费：当队列满时，生产者需要等待；当队列空时，消费者需要等待，以此避免资源浪费和无效操作。
提高效率：通过设置高水位线和低水位线，适时唤醒多个生产者或消费者，提高生产和消费的效率。

阻塞队列的实现

下面是一个基于C++的阻塞队列实现：

#ifndef __BLOCK_QUEUE__
#define __BLOCK_QUEUE__

#include <cstdlib>
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <queue>

using std::cout;
using std::endl;

template <class T>
class BlockQueue
{
private:
    bool Isfull()
    {
        return _queue.size() == _maxcap;
    }

    bool Isempty()
    {
        return _queue.empty();
    }

public:
    BlockQueue(int maxcap = 10)
        : _maxcap(maxcap), _queue(), _l_water(0), _h_water(maxcap), _call_num(maxcap / 2)
    {
        if (maxcap <= 4)
        {
            _l_water = 0x3f3f3f3f;
            _h_water = -0x3f3f3f3f;
            _call_num = 1;
        }
        else
        {
            _l_water = maxcap / 4;
            _h_water = maxcap * 3 / 4;
        }

        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_p_cond, nullptr);
        pthread_cond_init(&_c_cond, nullptr);
    }

    void Push(T &in)
    {
        pthread_mutex_lock(&_mutex);
        while (Isfull())
        {
            pthread_cond_wait(&_p_cond, &_mutex);
        }
        _queue.push(in);
        pthread_cond_signal(&_c_cond);
        if (_queue.size() >= _h_water)
        {
            cout << "call lots comsumer" << endl;
            int t = _call_num;
            while (t--)
                pthread_cond_signal(&_c_cond);
        }
        pthread_mutex_unlock(&_mutex);
    }

    void Pop(T *out)
    {
        pthread_mutex_lock(&_mutex);
        while (Isempty())
        {
            pthread_cond_wait(&_c_cond, &_mutex);
        }
        *out = _queue.front();
        _queue.pop();
        pthread_cond_signal(&_p_cond);
        if (_queue.size() <= _l_water)
        {
            cout << "call lots creator" << endl;
            int t = _call_num;
            while (t--)
                pthread_cond_signal(&_p_cond);
        }
        pthread_mutex_unlock(&_mutex);
    }

    ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_p_cond);
        pthread_cond_destroy(&_c_cond);
    }

    size_t GetSize()
    {
        return _queue.size();
    }

private:
    std::queue<int> _queue;
    int _maxcap;
    pthread_mutex_t _mutex;
    pthread_cond_t _p_cond;
    pthread_cond_t _c_cond;
    int _l_water;
    int _h_water;
    int _call_num;
};

#endif