在并发编程中，生产者-消费者 (Producer-Consumer) 模型是最经典的设计模式之一。它通过一个缓冲区 (Buffer) 将数据的生产和处理解耦，有效平衡了线程间的速度差异。经典的x客的WebServer服务器就有讲解到生产者-消费者模型，但是是用c语言实现的，这里用C++11实现一下，就当练手。

【Q1】std::lock_guard 和 std::unique_lock 怎么选？

• lock_guard：构造时上锁，析构时解锁，中途不能手动解锁。适合简单的临界区。
• unique_lock：支持手动上锁和解锁，比较灵活。配合condition_variable::wait必须用它，因为wait需要在挂起前释放锁，醒来后需要重新获得锁。

【Q2】emplace_back为什么能接收不同数量的参数？

• 它利用了C++11的变长参数模板，将参数包完美的转发给std::thread的构造函数，从而支持任意签名的任务函数。

【Q3】什么是虚假唤醒？怎么解决虚假唤醒？

• 在多线程环境中，线程挂起在condition_variable上等待。当它被唤醒（从wait函数返回）时，并不意味着条件一定满足了。

  if (queue_.empty()) {
      not_empty_.wait(lock); // 1. 释放锁并等待 -> 2. 被唤醒 -> 3. 重新拿锁
  }
  // 4. 醒来后直接执行
  queue_.pop();
  // 5.此时如果其他线程把队列里唯一的数据pop了，这里执行就会出错

• 把if换成while，在线程醒来后，还会对队列是否为空进行判断。

  while (queue_.empty()) {
      not_empty_.wait(lock);
  }
  // 只有当 queue_.empty() 为 false，才会跳出循环往下执行
  queue_.pop();

• 用lambda表达式和wait带有谓词的重载版本，可以直接实现while循环。

  // 语义更清晰：一直等，直到队列不为空为止
  not_empty_.wait(lock, [this](){ return !queue_.empty(); });

完整代码：

#include <iostream>
#include <mutex>
#include <thread>
#include <string>
#include <queue>
#include <condition_variable>
#include <chrono>

//线程安全的阻塞队列
template <typename T>
class BlockingQueue{
public:
    //这里explicit的作用防止隐式的类型转换
    explicit BlockingQueue(size_t capacity):capacity_(capacity){

    }
    //为什么禁止拷贝构造和赋值?
    //因为锁和条件变量是不允许被拷贝的。
    BlockingQueue(const BlockingQueue&)=delete;
    BlockingQueue& operator=(const BlockingQueue&)=delete;

    //生产者调用：将元素推入队列
    void Push(const T& value){
        std::unique_lock<std::mutex> lock(mutex_);

        //这里使用lambda表达式防止虚假唤醒
        //只有当队列大小小于容量时继续，否则等待
        not_full_.wait(lock, [this](){return queue_.size() < capacity_;});

        queue_.push(value);
        //唤醒一个消费者
        not_empty_.notify_one();
    }

    //消费者调用：从队列里取出元素
    T Pop(){
        std::unique_lock<std::mutex> lock(mutex_);

        not_empty_.wait(lock, [this](){return !queue_.empty();});
        T value = queue_.front();
        queue_.pop();

        not_full_.notify_one();

        return value;
    }
private:
    std::queue<T> queue_;
    size_t capacity_;

    std::mutex mutex_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
};

void ThreadSafePrint(const std::string& msg){
    //这里会定义一个静态的锁
    static std::mutex PrintMutex;
    //lock_guard和unique_lock的区别是什么？
    std::lock_guard<std::mutex> lock(PrintMutex);
    std::cout<< msg <<std::endl;
}

//生产者模型
void ProducerTask(BlockingQueue<int> &queue, int id, int count){
    for(int i=0; i<count; i++){
        int value = id*1000+i;

        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        queue.Push(value);

        ThreadSafePrint("Producer ["+std::to_string(id)+"] pushed:" +std::to_string(value));
    }
}

//消费者模型
void ConsumerTask(BlockingQueue<int> &queue, int id){
    while(true){
        std::this_thread::sleep_for(std::chrono::milliseconds(20));

        int value = queue.Pop();
        //收到毒药丸 结束信号
        if(value == -1){
            ThreadSafePrint("Consumer ["+std::to_string(id)+"] shutting down.");
            break;
        }

        ThreadSafePrint("Consumer ["+std::to_string(id)+"] popped: "+ std::to_string(value));
    }
}

int main(){
    const int kQueueCapacity = 5;
    const int kProducerCount = 2;
    const int kConsumerCount = 2;
    const int kItemsPerProducer = 100;

    BlockingQueue<int> queue(kQueueCapacity);
    std::vector<std::thread> producers;
    std::vector<std::thread> consumers;

    //启动生产者
    for(int i=0; i<kProducerCount; ++i){
        producers.emplace_back(ProducerTask, std::ref(queue), i+1, kItemsPerProducer);
    } 

    //启动消费者
    for(int i=0; i<kConsumerCount; ++i){
        consumers.emplace_back(ConsumerTask, std::ref(queue),i+1);
    }

    for(auto &p:producers){
        p.join();
    }

    for(int i=0; i<kConsumerCount; i++){
        queue.Push(-1);
    }

    for(auto& c:consumers){
        c.join();
    }

    ThreadSafePrint("All tasks finished.");
    return 0;
}