c++11 - C++11中的并发阻塞队列

Question

对于线程之间的消息传递，我正在寻找具有以下属性的并发队列：

• 有界大小
• 阻塞/等待直到元素可用的 pop 方法。
• abort 方法取消等待
• 可选：优先级

多个生产者，一个消费者。

TBB将concurrent_bounded_queue提供这一点，但我正在寻找替代方案以避免对 TBB 的额外依赖。

该应用程序使用 C++11 和 boost。我找不到任何适合提升的东西。有什么选择？

Answer 1

使用 Boost 库（循环缓冲区）和 C++11 标准库的幼稚实现。

#include <mutex>
#include <condition_variable>
#include <boost/circular_buffer.hpp>

struct operation_aborted {};

template <class T, std::size_t N>
class bound_queue {
public:
  typedef T value_type;
  bound_queue() : q_(N), aborted_(false) {}
  void push(value_type data)
  {
    std::unique_lock<std::mutex> lk(mtx_);
    cv_pop_.wait(lk, [=]{ return !q_.full() || aborted_; });
    if (aborted_) throw operation_aborted();
    q_.push_back(data);
    cv_push_.notify_one();
  }
  value_type pop()
  {
    std::unique_lock<std::mutex> lk(mtx_);
    cv_push_.wait(lk, [=]{ return !q_.empty() || aborted_; });
    if (aborted_) throw operation_aborted();
    value_type result = q_.front();
    q_.pop_front();
    cv_pop_.notify_one();
    return result;
  }
  void abort()
  {
    std::lock_guard<std::mutex> lk(mtx_);
    aborted_ = true;
    cv_pop_.notify_all();
    cv_push_.notify_all(); 
  }
private:
  boost::circular_buffer<value_type> q_;
  bool aborted_;
  std::mutex mtx_;
  std::condition_variable cv_push_;
  std::condition_variable cv_pop_;
};