c++ - 生产者消费者线程关闭 c++

Question

您好，给出以下 c++11 代码的快速问题，它适用于生产者/消费者，问题是我想关闭 DataQueue 并停止所有消费者。尽管存在问题，因为问题在于消费者只是调用 popWait() 并且可以被阻止。在这种情况下，我该如何关闭我的消费者？这可能是一个需要纠正的设计问题。我试图不招致任何性能损失，因为这段代码理想情况下应该使用中断模式或类似模式来释放队列锁。尽管如此，我想知道是否有什么简单的方法可以让消费者知道在生产者关闭时停止调用 pop 等待函数。棘手的部分是如果队列中仍有数据则关闭，您将不得不等待消费者完成提取数据。我相信我有一个解决方案，消费者有自己的关闭但对想法持开放态度。提前致谢。

#ifndef __DataQueue_h__
#define __DataQueue_h__

#include <mutex>
#include <queue>
#include <condition_variable>
#include <chrono>

template <typename DataT>
class DataQueue
{
  public:

  DataQueue (): _shutdown(false), _waitTime(5), _itemAvailable() {}

  void push ( const DataT& data )       
  {
    std::unique_lock<std::mutex> lock(_mutex);
    queue.push(data);
    _itemAvailable.notify_one();
  }

// worked fine until I need to shutdown services... then some were blocked
  DataT popWait()
  {
    std::unique_lock<std::mutex> lock(_mutex);

    if(queue.empty())
    {
      _itemAvailable.wait(lock);
    }

    DataT temp(queue.front());
    queue.pop();

    return temp;
  }

  inline void shutdown()
  {
    _shutdown = true;
  }

  private:
  std::queue<DataT> queue;
  bool _shutdown;
  unsigned int _waitTime;
  std::mutex _mutex;
  std::condition_variable _itemAvailable;

};

#endif

Answer 1

一个想法是在调用 shutdown 时唤醒所有消费者。 ¹。在popWait方法，然后您可以在从 wait 返回时检查是否设置了关闭标志.

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

template <typename DataT>
class DataQueue
{
public:
    DataQueue (): _shutdown(false), _itemAvailable() {}

    void push ( const DataT& data )       
    {
        std::unique_lock<std::mutex> lock(_mutex);
        queue.push(data);
        _itemAvailable.notify_one();
    }

    Maybe<DataT> popWait()
    {
        std::unique_lock<std::mutex> lock(_mutex);

        while(queue.empty() && !_shutdown)
        {
          _itemAvailable.wait(lock);
        }

        Maybe<DataT> data;
        // leave pending data in the queue
        if (_shutdown) 
        {
            // consumers should stop polling when receiving an 'empty' value
            return data;
        }

        data.add(queue.front());
        queue.pop();   
        return data;
    }

    inline void shutdown()
    {
        _shutdown = true;
        _itemAvailable.notify_all();
    }

private:
    std::queue<DataT> queue;
    std::atomic<bool> _shutdown;
    std::mutex _mutex;
    std::condition_variable _itemAvailable;
};

popWait 的返回值

除了所有同步和信号的东西，你还必须重新考虑 popWait 的返回值.如果你想实现一个通用的 shutdown()方法，即不将特殊标记值填充到队列本身，popWait必须能够返回一个“值”，表明生产者已经停止——可能是类似 Maybe<DataT> 的模板。 ²。我设想 Maybe<DataT>可以返回 DataT或什么都没有，在这种情况下，消费者将停止轮询。

template<typename DataT>
class Maybe 
{
   DataT _data;
   bool _empty;

pulic:
   Maybe() : _data(), _empty(true) {};

   void add(const DataT& raData)
   {
      _data=raData;
      _empty=false;
   }

   bool isEmpty() const
   {
      return _empty;
   }

   DataT get() const
   {
      return _data;
   }
}

这是一个相当原始的界面。您可以根据需要扩展它。

¹ ComicSansMS向我指出我应该声明 _shutdown成员变量为 std::atmic<bool>以避免内存重新排序问题。感谢您的提醒。

²我偶然发现了 std::optional<T> (C++14 中的新增功能)，这基本上就是我的想法。