c++ - 我的生产者消费者队列的任何明显问题或改进

Question

我问了一个previous question关于过于笼统的生产者/消费者代码(尽管答案肯定有帮助)。所以我采纳了 earlier SO question 的建议由另一位作者将它们转换为 C++ 并 boost 。然而，我总是有点担心多线程代码 - 所以如果有人能看到任何明显的改进，我很乐意听到它们。

#include <pthread.h>
#include <deque>
#include <iostream>

#include "boost/thread.hpp"


class MyQueue
{
protected:
  boost::mutex mutex_;
  boost::condition_variable condition_;
  bool cancel_;
  std::deque<int> data_;

public:
  MyQueue() : mutex_(), condition_(), cancel_(false), data_()
  {
  }

  struct Canceled{};

  void push( int i )
  {
     boost::lock_guard<boost::mutex> l(mutex_);
     if(cancel_) throw Canceled();
     data_.push_back(i);
     condition_.notify_all();
  }

  void pop( int & i )
  {
     boost::unique_lock<boost::mutex> l(mutex_);
     while(! cancel_ && data_.size()==0 )
     {
        condition_.wait( l );
     }
     if(cancel_) throw Canceled();

     assert( data_.size() != 0 );
     i = data_.front();
     data_.pop_front();
  }

  void cancel()
  {
     boost::lock_guard<boost::mutex> l(mutex_);
     if( cancel_) throw Canceled();
     cancel_ = true;
     condition_.notify_all();
  }
};


boost::mutex iomutex;

void producer( MyQueue * q, const std::string & name )
try
{
  for(unsigned int i=0 ; i<20; ++i)
  {
    q->push( i );
    boost::lock_guard<boost::mutex> l(iomutex);
    std::cout<<name<<"  PRODUCED "<<i<<std::endl;
  }

  sleep(1);
  q->cancel();
  {
    boost::lock_guard<boost::mutex> l(iomutex);
    std::cout<<name<<"  PRODUCER EXITING NORMALLY"<<std::endl;
  }
}
catch( MyQueue::Canceled & c )
{
  boost::lock_guard<boost::mutex> l(iomutex);
  std::cout<<name<<"  PRODUCER CANCLED "<<std::endl;
}

void consumer( MyQueue * q, const std::string & name )
try
{
  while(true)
  {
    int i;
    q->pop( i );
    boost::lock_guard<boost::mutex> l(iomutex);
    std::cout<<name<<"  CONSUMED "<<i<<std::endl;
  }
}
catch( MyQueue::Canceled & c )
{
  boost::lock_guard<boost::mutex> l(iomutex);
  std::cout<<name<<"  CONSUMER CANCLED "<<std::endl;
}

int main()
{
  MyQueue q;
  boost::thread pr1( producer, &q, "pro1");
  boost::thread pr2( producer, &q, "pro2");
  boost::thread cons1( consumer, &q, "con1");
  boost::thread cons2( consumer, &q, "con2");

  pr1.join();
  pr2.join();
  cons1.join();
  cons2.join();
}

更新:我最终使用了 Anthony Williams' concurrent queue 的修改版本.我的修改版本可以找到here .

Answer 1

如果您担心实现中的潜在缺陷，您可以尝试使用 Anthony Williams(Boost.Thread 库的维护者)出色的线程安全，multiple-producer, multiple-consumer queue .