C++中的多线程队列销毁

问题描述：

所以我有一个共享并发队列。它似乎很好地工作，除了销毁。C++中的多线程队列销毁

队列实现的方式是它包含一个条件变量和互斥对。几个工作线程启动，等待这个条件变量。当新对象可用于处理时，它们被推入队列并且条件变量被发送。

问题是，当主线程退出时，销毁队列，条件变量将被销毁，但是当条件变量正在使用时会失败。这引发了一个异常，并且所有事情都很快就爆发了。

我想指示工人，将他们唤醒并让他们退出，等待他们完成，然后继续执行主线程。我的问题是当这些线程完成时 - 我是否需要额外的同步原语？

反正继承人队列代码：

// Based on code from http://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html 
// Original version by Anthony Williams 
// Modifications by Michael Anderson 

#include "boost/thread.hpp" 
#include <deque> 

template<typename Data> 
class concurrent_queue 
{ 
private: 
    std::deque<Data> the_queue; 
    mutable boost::mutex the_mutex; 
    boost::condition_variable the_condition_variable; 
    bool is_canceled; 

public: 
    concurrent_queue() : the_queue(), the_mutex(), the_condition_variable(), is_canceled(false) {} 
    struct Canceled{}; 
    void push(Data const& data) 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 
     if (is_canceled) throw Canceled(); 
     the_queue.push_back(data); 
     lock.unlock(); 
     the_condition_variable.notify_one(); 
    } 

    bool empty() const 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 
     if (is_canceled) throw Canceled(); 
     return the_queue.empty(); 
    } 

    bool try_pop(Data& popped_value) 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 
     if (is_canceled) throw Canceled(); 
     if(the_queue.empty()) 
     { 
      return false; 
     } 

     popped_value=the_queue.front(); 
     the_queue.pop_front(); 
     return true; 
    } 

    void wait_and_pop(Data& popped_value) 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 

     while(the_queue.empty() && !is_canceled) 
     { 
      the_condition_variable.wait(lock); 
     } 
     if (is_canceled) throw Canceled(); 

     popped_value=the_queue.front(); 
     the_queue.pop_front(); 
    } 

    std::deque<Data> wait_and_take_all() 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 

     while(the_queue.empty() && !is_canceled) 
     { 
      the_condition_variable.wait(lock); 
     } 
     if (is_canceled) throw Canceled(); 

     std::deque<Data> retval; 
     std::swap(retval, the_queue); 
     return retval; 
    } 

    void cancel() 
    { 
     boost::mutex::scoped_lock lock(the_mutex); 
     if (is_canceled) throw Canceled(); 
     is_canceled = true; 
     lock.unlock(); 
     the_condition_variable.notify_all(); 
    } 

};

答

您可以在每个线程调用join()等待，直到它已完成执行。类似这样的：

void DoWork() {}; 

int main() 
{ 
    boost::thread t(&DoWork); 
    // signal for the thread to exit 
    t.join(); // wait until it actually does exit 

    // destroy the queue 
}

或者您可以使用boost::thread_group多个线程。

int main() 
{ 
    boost::thread_group tg; 

    for(int i = 0 ; i < 10 ; ++i) 
     tg.create_thread(&DoWork); 

    // signal to stop work 

    tg.join_all(); 

    // destroy the queue 
}

其他线程都在等待上condition_variable ..他们仍然需要被唤醒，然后加入。这都需要排队析构函数，然后意味着队列需要跟踪的所有它的工作..也许使用升压线程内发生：thread_group会提出这样的一个小整洁，虽然。 – 2011-02-05 04:37:56

它看起来像有一个`cancel（）`函数可以唤醒每个线程并抛出'Cancelled`异常。然后，每个线程都可以捕获`Cancelled`异常并调用`return`。我误解了你写的'cancel（）'吗？ – JaredC 2011-02-05 04:44:46

答

您有两种选择。当队列超出范围时，实际上并不会在其他线程引用时被销毁（即使用shared_ptr，将其传递给其他线程;在main（）的末尾调用cancel（）;一旦其他线程抛出已取消并可能退出，队列将被破坏）。

或者，如果你想确保它的实际的main（）的结束时被销毁，那么你就需要等待其他线程。如果您在处理析构函数之外的等待时可以执行JaredC的建议。要在析构函数内部完成它，看起来更清晰的是不要存储所有的线程，而只是为了保留一个计数和另一个同步原语。无论哪种方式，您都需要队列来维持某种状态，以等待所有线程完成。

对我来说，似乎第一个解决方案（与shared_ptr）更清洁。

C++中的多线程队列销毁

相关推荐