生产者消费者模型

阻塞队列

用C++简易实现生产者消费者模型，即实现一个阻塞队列；

之前Web服务器项中的异步日志系统也用到了该模型；

---

• 生产者生产数据到缓冲区中，消费者从缓冲区中取数据；
• 缓冲区满时，生产者线程阻塞，进入等待状态。
• 缓冲区空时，消费者线程阻塞，进入等待状态。
• 生产者生产资源后，会唤醒可能等待着的消费者；
• 消费者消耗资源后，会唤醒可能阻塞着的生产者；

/*
 * @Description  : 封装生产者消费者模式
 * @Date         : 2022-10-12 19:29:22
 * @LastEditTime : 2022-10-12 20:10:32
 */
#include <bits/stdc++.h>

/**
 * @description: 阻塞队列类
 */
template <class T>
class BlockQueue {
private:
    std::queue<T> que_;       // 共享队列
    std::mutex    mtx_;       // 互斥量
    size_t        capacity_;  // 最大容量

    std::condition_variable cv_consumer_;  // 消费者条件变量
    std::condition_variable cv_producer_;  // 生产者条件变量

public:
    explicit BlockQueue(size_t MaxCapacity = 100) : capacity_(MaxCapacity) {
        assert(MaxCapacity > 0);
    }

    void push(const T &item);
    T    pop();
};

/**
 * @description: 向队尾加入一个元素
 */
template <typename T>
void BlockQueue<T>::push(const T &item) {
    std::unique_lock<std::mutex> locker(mtx_);

    /*条件变量控制，如果元素超过了容量，需要阻塞等待消费者消费后才能加入*/
    while (que_.size() >= capacity_) {
        /*生产者等待*/
        cv_producer_.wait(locker);
    }

    que_.push(item);

    /*唤醒一个消费者*/
    cv_consumer_.notify_one();
}

/**
 * @description: 在队头弹出一个元素
 */
template <typename T>
T BlockQueue<T>::pop() {
    std::unique_lock<std::mutex> locker(mtx_);

    /*如果队列为空，消费者条件变量等待*/
    while (que_.empty()) {
        cv_consumer_.wait(locker);
    }

    T item = que_.front();
    que_.pop();

    /*唤醒一个生产者*/
    cv_producer_.notify_one();

    return item;
}

void producer(BlockQueue<int> &q) {
    for (int i = 0; i < 10; i++) {
        q.push(i);
        std::cout << "push: " << i << std::endl;
    }
}

void comsumer(BlockQueue<int> &q) {
    for (int i = 0; i < 10; i++) {
        int item = q.pop();
        std::cout << "pop:" << item << std::endl;
    }
}

int main() {
    BlockQueue<int> bq(5);

    std::thread t1(producer, std::ref(bq));
    std::thread t2(comsumer, std::ref(bq));

    t1.join();
    t2.join();

    return 0;
}

编译用：

g++ BlockQueue.cpp -o bq -pthread