停止等待std :: condition_variable的C++ 11 std :: threads

Question

我试图理解新的C++ 11标准中的基本多线程机制.我能想到的最基本的例子如下:

• 生产者和消费者在不同的线程中实现
• 生产者在队列中放置一定数量的项目
• 如果存在任何存在,则消费者从队列中获取项目

这个例子也被用在许多关于多线程的学校书籍中,关于通信过程的一切都很好.但是,在停止使用者线程时我遇到了问题.

我希望消费者运行直到它获得明确的停止信号(在大多数情况下,这意味着我等待生产者完成所以我可以在程序结束之前停止消费者).不幸的是,C++ 11线程缺少一种中断机制(例如我从Java中的多线程中知道).因此,我必须使用标志isRunning来表示我想要一个线程停止.

现在的主要问题是:在我停止生产者线程之后,队列为空并且消费者在等待a condition_variable再次填充队列时获取信号.所以我需要通过notify_all()在退出之前调用变量来唤醒线程.

我找到了一个有效的解决方案,但似乎有些混乱.下面列出了示例代码(很抱歉,但不知何故,我无法为"最小"最小示例减少代码大小):

队列类:

class Queue{
public:
    Queue() : m_isProgramStopped{ false } { }

    void push(int i){
        std::unique_lock<std::mutex> lock(m_mtx);
        m_q.push(i);
        m_cond.notify_one();
    }

    int pop(){
        std::unique_lock<std::mutex> lock(m_mtx);
        m_cond.wait(lock, [&](){ return !m_q.empty() || m_isProgramStopped; });

        if (m_isProgramStopped){
            throw std::exception("Program stopped!");
        }

        int x = m_q.front();
        m_q.pop();

        std::cout << "Thread " << std::this_thread::get_id() << " popped " << x << "." << std::endl;
        return x;
    }

    void stop(){
        m_isProgramStopped = true;
        m_cond.notify_all();
    }

private:
    std::queue<int> m_q;
    std::mutex m_mtx;
    std::condition_variable m_cond;
    bool m_isProgramStopped;
};

制片人:

class Producer{
public:
    Producer(Queue & q) : m_q{ q }, m_counter{ 1 } { }

    void produce(){
        for (int i = 0; i < 5; i++){
            m_q.push(m_counter++);
            std::this_thread::sleep_for(std::chrono::milliseconds{ 500 });
        }
    }

    void execute(){
        m_t = std::thread(&Producer::produce, this);
    }

    void join(){
        m_t.join();
    }

private:
    Queue & m_q;
    std::thread m_t;

    unsigned int m_counter;
};

消费者:

class Consumer{
public:
    Consumer(Queue & q) : m_q{ q }, m_takeCounter{ 0 }, m_isRunning{ true }
    { }

    ~Consumer(){
        std::cout << "KILL CONSUMER! - TOOK: " << m_takeCounter << "." << std::endl;
    }

    void consume(){
        while (m_isRunning){
            try{
                m_q.pop();
                m_takeCounter++;
            }
            catch (std::exception e){
                std::cout << "Program was stopped while waiting." << std::endl;
            }
        }
    }

    void execute(){
        m_t = std::thread(&Consumer::consume, this);
    }

    void join(){
        m_t.join();
    }

    void stop(){
        m_isRunning = false;
    }

private:
    Queue & m_q;
    std::thread m_t;

    unsigned int m_takeCounter;
    bool m_isRunning;
};

最后main():

int main(void){
    Queue q;

    Consumer cons{ q };
    Producer prod{ q };

    cons.execute();
    prod.execute();

    prod.join();

    cons.stop();
    q.stop();

    cons.join();

    std::cout << "END" << std::endl;

    return EXIT_SUCCESS;
}

这是结束等待条件变量的线程的正确方法还是有更好的方法？目前,队列需要知道程序是否已经停止(在我看来,这会破坏组件的松散耦合),我需要stop()明确地调用队列,这似乎是正确的.

另外,如果队列为空,那么条件变量应该只用作信号,现在代表另一个条件 - 如果程序已经结束.如果我没有弄错,每次线程等待条件变量发生某些事件时,它还必须检查线程是否必须在继续执行之前停止(这似乎也是错误的).

我是否有这些问题,因为我的整个设计都有问题,或者我错过了一些可以用来以干净的方式退出线程的机制？

Answer 1

不,您的设计没有任何问题,这是针对此类问题采取的常规方法.

对条件变量附加多个条件(例如队列或程序停止的任何条件)完全有效.关键是在wait返回时检查条件中的位.

没有标志Queue来指示程序正在停止,你应该将标志视为"我能接受".这是一个更好的整体范例,在多线程环境中工作得更好.

此外,pop如果有人调用它并且stop已被调用,则可以替换在bool try_pop(int &value)返回true值时将返回的方法,否则不会抛出异常false.这样调用者可以检查是否已经停止查看队列(添加bool is_stopped() const方法).虽然异常处理在这里起作用,但它有点沉重,并不是多线程程序中的特殊情况.