C ++相当于Java的BlockingQueue

Posted 2021-03-27

我正在将一些Java代码移植到C ++，并且一个特定部分使用BlockingQueue将消息从许多生产者传递给单个消费者。

如果您不熟悉Java BlockingQueue是什么，它只是一个具有硬容量的队列，它将线程安全方法暴露给队列中的put（）和take（）。如果队列已满，则put（）阻塞;如果队列为空，则使用take（）块。此外，还提供了这些方法的超时敏感版本。

超时与我的用例相关，因此提供这些超时的建议是理想的。如果没有，我可以自己编写代码。

我已经google了一下，并迅速浏览了Boost库，我找不到这样的东西。也许我在这里失明了......但有人知道一个好推荐吗？

谢谢！

答案

它不是固定大小，它不支持超时，但这是我最近使用C ++ 2011构造发布的队列的简单实现：

#include <mutex>
#include <condition_variable>
#include <deque>

template <typename T>
class queue
{
private:
    std::mutex              d_mutex;
    std::condition_variable d_condition;
    std::deque<T>           d_queue;
public:
    void push(T const& value) {
        {
            std::unique_lock<std::mutex> lock(this->d_mutex);
            d_queue.push_front(value);
        }
        this->d_condition.notify_one();
    }
    T pop() {
        std::unique_lock<std::mutex> lock(this->d_mutex);
        this->d_condition.wait(lock, [=]{ return !this->d_queue.empty(); });
        T rc(std::move(this->d_queue.back()));
        this->d_queue.pop_back();
        return rc;
    }
};

扩展和使用定时等待弹出应该是微不足道的。我没有做到的主要原因是我对目前为止我想到的界面选择不满意。

另一答案

以下是blocking queue with shutdown request功能的示例：

template <typename T> class BlockingQueue {
  std::condition_variable _cvCanPop;
  std::mutex _sync;
  std::queue<T> _qu;
  bool _bShutdown = false;

public:
  void Push(const T& item)
  {
    {
      std::unique_lock<std::mutex> lock(_sync);
      _qu.push(item);
    }
    _cvCanPop.notify_one();
  }

  void RequestShutdown() {
    {
      std::unique_lock<std::mutex> lock(_sync);
      _bShutdown = true;
    }
    _cvCanPop.notify_all();
  }

  bool Pop(T &item) {
    std::unique_lock<std::mutex> lock(_sync);
    for (;;) {
      if (_qu.empty()) {
        if (_bShutdown) {
          return false;
        }
      }
      else {
        break;
      }
      _cvCanPop.wait(lock);
    }
    item = std::move(_qu.front());
    _qu.pop();
    return true;
  }
};

另一答案

BlockingCollection是一个C ++ 11线程安全集合类，它以.NET BlockingCollection类为模型。

它支持以下内容：

• 经典生产者/消费者模式的实施（即条件变量，互斥体）
• 从多个线程并发添加和获取项目。
• 可选的最大容量。
• 收集为空或满时阻止的插入和删除操作。
• 插入和删除“尝试”不阻止或阻止达到指定时间段的操作。
• 插入和删除“批量”操作，允许一次添加或获取多个元素。
• FIFO，LIFO，基于优先级的插入和删除操作。
• 通过管理生产者和消费者线程的活动子集，最大限度地减少睡眠，唤醒和锁定争用。
• 基于范围的循环支持。