C++线程安全队列关闭

Question

我在 C++ 中使用这个类来设置生产者-消费者：

#pragma once

#include <queue>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <atomic>

template <typename T> class SafeQueue
{
public:
    SafeQueue() :
    _shutdown(false)
    {

    }

    void Enqueue(T item)
    {
        std::unique_lock<std::mutex> lock(_queue_mutex);
        bool was_empty = _queue.empty();
        _queue.push(std::move(item));
        lock.unlock();

        if (was_empty)
            _condition_variable.notify_one();
    }

    bool Dequeue(T& item)
    {
        std::unique_lock<std::mutex> lock(_queue_mutex);

        while (!_shutdown && _queue.empty())
            _condition_variable.wait(lock);

        if(!_shutdown)
        {
            item = std::move(_queue.front());
            _queue.pop();

            return true;
        }

        return false;
    }

    bool IsEmpty()
    {
        std::lock_guard<std::mutex> lock(_queue_mutex);
        return _queue.empty();
    }

    void Shutdown()
    {
        _shutdown = true;
        _condition_variable.notify_all();
    }

private:
    std::mutex _queue_mutex;
    std::condition_variable _condition_variable;
    std::queue<T> _queue;
    std::atomic<bool> _shutdown;
};

我是这样使用它的：

class Producer
{
public:
    Producer() :
        _running(true),
        _t(std::bind(&Producer::ProduceThread, this))
    { }

    ~Producer()
    {
        _running = false;
        _incoming_packets.Shutdown();
        _t.join();
    }

    SafeQueue<Packet> _incoming_packets;

private:
    void ProduceThread()
    {
        while(_running)
        {
            Packet p = GetNewPacket();
            _incoming_packets.Enqueue(p);
        }
    }

    std::atomic<bool> _running;
    std::thread _t;
}

class Consumer
{
    Consumer(Producer* producer) :
        _producer(producer),
        _t(std::bind(&Consumer::WorkerThread, this))
    { }

    ~Consumer()
    {
        _t.join();
    }

private:
    void WorkerThread()
    {
        Packet p;

        while(producer->_incoming_packets.Dequeue(p))
            ProcessPacket(p);
    }

    std::thread _t;
    Producer* _producer;
}

这大部分都有效。但是偶尔当我删除生产者（并导致它的解构器调用SafeQueue::Shutdown，_t.join() 永远阻塞。

我的猜测是问题出在此处（SafeQueue::Dequeue）：

while (!_shutdown && _queue.empty())
        _condition_variable.wait(lock);

来自线程#1 的SafeQueue::Shutdown 在线程#2 完成检查_shutdown 时被调用，但在它执行_condition_variable.wait(lock) 之前，它“错过”了notify_all()。这会发生吗？

如果这是问题所在，最好的解决方法是什么？

Answer 1

由于 SafeQueue 对象归生产者所有，因此删除生产者会导致通知消费者和在 ~Producer 完成时从其下删除 SafeQueue 之间的竞争条件。

我建议共享资​​源既不属于生产者也不属于消费者，而是作为参考传递给每个人的构造函数。

更改生产者和消费者构造函数；

Producer( SafeQueue<Packet> & queue ) :
    _running(false), _incoming_packets(queue) {}


Consumer( SafeQueue<Packet> & queue ) :
    _running(false), _incoming_packets(queue) {}

以这种方式使用您的实例；

SafeQueue<Packet> queue;
Producer producer(queue);  
Consumer consumer(queue);

...do stuff...

queue.shutdown();

这也解决了 Consumer 类与 Producer 类紧密耦合的糟糕设计问题。

另外，在析构函数中杀死和加入线程可能是个坏主意，就像为 ~Producer 所做的那样。最好为每个线程类添加一个 Shutdown() 方法，并显式调用它们；

producer.shutdown();
consumer.shutdown();
queue.shutdown();

关闭顺序并不重要，除非您担心在停止消费者时会丢失仍在队列中的未处理数据包。

Answer 2

在您的SafeQueue::Dequeue 中，您可能以错误的方式使用std::condition_variable...更改此：

bool Dequeue(T& item)
{
    std::unique_lock<std::mutex> lock(_queue_mutex);

    while (!_shutdown && _queue.empty())
        _condition_variable.wait(lock);
    if(!_shutdown)
    {
        item = std::move(_queue.front());
        _queue.pop();
        return true;
    }
    return false;
}

到

bool Dequeue(T& item)
{
    std::unique_lock<std::mutex> lock(_queue_mutex);
   _condition_variable.wait(lock, []{ return _shutdown || !_queue.empty() });
    if(!_shutdown)
    {
        item = std::move(_queue.front());
        _queue.pop();
        return true;
    }

    return false;
}

其次，Consumer中数据成员的初始化顺序与其构造函数有关

class Consumer
{
    Consumer(Producer* producer) :
        _producer(producer),
        _t(std::bind(&Consumer::WorkerThread, this))
    { }
    ......
    // _t will be constructed first, regardless of your constructor initializer list
    // Meaning, the thread can even start running using an unintialized _producer
    std::thread _t;      
    Producer* _producer;
}

应该重新排序：

class Consumer
{
    Consumer(Producer* producer) :
        _producer(producer),
        _t(std::bind(&Consumer::WorkerThread, this))
    { }
    ......
    Producer* _producer;
    std::thread _t;      
}

CAB's answer 涵盖了您的问题的另一部分