在无限循环中唤醒等待条件的线程

Question

我有一个非常基本的生产者/消费者实现。生产者是“主”线程，消费者在单独的线程上执行。但是，需要使用 Start() 函数显式启动消费者。这会将“处理”标志设置为 true（在无限 while 循环中使用）。

一旦进入while循环，消费者就会使用条件变量来查看队列中是否有数据需要处理。如果是，它会完成它的工作，回到无限循环的顶部，然后是条件变量，等等。

我遇到的问题是消费者正在等待队列中的数据，我想停止处理。 如何唤醒消费者？我在下面提供了一些示例代码，删除了一些主要组件，只是展示了高级设计（实际上并不是所有内容都公开）。

// Consumer object
class Consumer {
public: 
  std::mutex mtx_;
  bool processing_ = false;
  std::thread processing_thread_;
  std::queue<int> data_;
  std::condition_variable cv_;
  ~Consumer() {
    // Make sure the processing thread is stopped
    {
      std::lock_guard<std::mutex> lock(mtx_);
      processing_ = false;
    }

    if (processing_thread_.joinable()) {

      processing_thread_.join();
    }
  }
  void Start() {
    std::lock_guard<std::mutex> lock(mtx_);

    processing_ = true;
    processing_thread_ = std::thread(
      &Consumer::Run,
       this);
  }
  void Stop() {
    std::lock_guard<std::mutex> lock(mtx_);

    processing_ = false;
  } 
  void AddData(int d) {
    std::lock_guard<std::mutex> lock(mtx_);
    data_.push(d);
    cv_.notify_one();
  }
  bool IsDataAvailable() const {
    return (!data.empty());
  }
  void Run() {

    // The infinite loop
    while (processing_) {
      // This is where I get stuck waiting even tho processing has been 
      // changed to false by the main thread
      std::unique_lock<std::mutex> lock(mtx_);
      cv_.wait(lock, std::bind(
        &Consumer::IsDataAvailable, this));
      // do some processing
    }
  }

}; // end of consumer

// Somewhere in main trying to stop the processing thread cause I am
// done processing OR my_consumer goes out of scope and tries to join
// ... 
  my_consumer.Stop();
}
// my_consumer goes out of scope here calling destructor.

Answer 1

消费者需要进行一些更改才能等待processing_中的更改：

~Consumer() {
  if (processing_thread_.joinable()) {
    Stop();
    processing_thread_.join();
  }
}
// ...
void Stop() {
  std::lock_guard<std::mutex> lock(mtx_);
  processing_ = false;
  cv_.notify_one();
} 
// ...

void Run() {
  for(;;) {
    std::unique_lock<std::mutex> lock(mtx_);
    // Wait till something is put into the queue or stop requested.
    cv_.wait(lock, [this]() { return !processing_ || !data_.empty(); });
    if(!data_.empty())
      // Process queue elements.
    else if(!processing_)
      return; // Only exit when the queue is empty.
  }
}