线程安全的队列关闭

我在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);

SafeQueue::Shutdown在线程#2完成检查_shutdown时被调用，但在执行_condition_variable.wait(lock)之前，因此它"错过"notify_all()。这种情况会发生吗?

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