阻塞队列的实现源码

刺骨的言语ヽ痛彻心扉 2024-03-25 23:18 40阅读 0赞

## 1.BlockQueue.hpp ##

#pragma once
    #include <iostream>
    #include <pthread.h>
    #include <mutex>
    #include <queue>
    #include <unistd.h>
    //STL容器存在线程安全问题
    
    const int gDefaultCap = 5;//最大容量capacity
    
    template<class T>
    class BlockQueue
    {
    private:
        bool isQueueEmpty()
        {
            return _bq.size() == 0;
        }
        bool isQueueFull()
        {
            return _bq.size() == _capacity;
        }
    public:
        BlockQueue(int capacity = gDefaultCap):_capacity(capacity)
        {
            pthread_mutex_init(&_mtx, nullptr);
            pthread_cond_init(&_isFull, nullptr);
            pthread_cond_init(&_isEmpty, nullptr);
        }
    
        void Push(const T& in)
        {
            pthread_mutex_lock(&_mtx);
            // 1.先检测当前临界资源是否满足临界条件
            // pthread_cond_wait:第二个参数是一把锁，成功调用wait之后，传入的锁会自动释放
            // 从哪里阻塞挂起，就从哪里唤醒
            // 唤醒时pthread_cond_wait会自动获取锁
            while(isQueueFull()) pthread_cond_wait(&_isFull, &_mtx);
            // 2.访问临界资源
            _bq.push(in);
            // if(_bq.size() >= _capacity/2) pthread_cond_signal(&_isEmpty);
            pthread_cond_signal(&_isEmpty);
            pthread_mutex_unlock(&_mtx);
        }
    
        void Pop(T* out)
        {
            pthread_mutex_lock(&_mtx);
            while(isQueueEmpty()) pthread_cond_wait(&_isEmpty, &_mtx);
    
            *out = _bq.front();
            _bq.pop();
            pthread_cond_signal(&_isFull);//安全的
            pthread_mutex_unlock(&_mtx);
        }
    
        ~BlockQueue()
        {
            pthread_mutex_destroy(&_mtx);
            pthread_cond_destroy(&_isEmpty);
            pthread_cond_destroy(&_isEmpty);
        }
    private:
        std::queue<T> _bq;       // 阻塞队列
        int _capacity;           // 容量上限
        pthread_mutex_t _mtx;    // 通过互斥锁保证队列安全
        pthread_cond_t _isEmpty; // 表示阻塞队列是否为空的条件
        pthread_cond_t _isFull;  // 表示阻塞队列是否为满的条件
    };

## 2.ConProd.cc ##

#include "BlockQueue.hpp"
    
    void* consumer(void *args)
    {
        BlockQueue<int> *bqueue = (BlockQueue<int> *)args;
        while(true)
        {
            int a;
            bqueue->Pop(&a);
            std::cout << "消费一个数据：" << a << std::endl;
            sleep(1);
        }
        return nullptr;
    }
    
    void* productor(void *args)
    {
        BlockQueue<int> *bqueue = (BlockQueue<int> *)args;
        int a = 1;
        while(true)
        {
            bqueue->Push(a);
            std::cout << "生产一个数据：" << a << std::endl;
            a++;
        }
        return nullptr;
    }
    
    int main()
    {
        BlockQueue<int> *bqueue = new BlockQueue<int>();
        pthread_t c, p; // pthread_create最后一个参数为传入的数据
        pthread_create(&c, nullptr, consumer, bqueue);
        pthread_create(&p, nullptr, productor, bqueue);
    
        pthread_join(c, nullptr);
        pthread_join(p, nullptr);
    
        delete bqueue;
        return 0;
    }