C++多线程实现生产者消费者模式运行后冻结无法退出如何解决？

问题描述 #

我为了学习尝试编写简单的生产者/消费者代码，由生产者将数字推入队列，消费者线程打印数字，以下是我的代码：

const int N_THREADS = 10;
const int N_TESTS = 100;
bool finished = false;
queue<int> q;

void produce()
{
    for (int i = 0; i < N_TESTS; i++)
    {
        q.push(i);
        cv.notify_all();        
    }
    finished = true;
}

void consume()
{
    while (!q.empty() || !finished)
    {   
        unique_lock<mutex> lock(m);
        cv.wait(lock, [] {return !q.empty(); });
        int i = q.front();
        cout << i << endl;
        q.pop();
    }   
}

int main()
{
    // 生产者线程
    thread producer(produce);
    
    // 消费者线程组
    vector<thread> consumers(N_THREADS);
    for (int i = 0; i < N_THREADS; i++)
    {
        consumers[i] = thread(consume);
    }

    // 等待所有线程执行结束
    producer.join();
    for (int i = 0; i < N_THREADS; i++)
    {
        consumers[i].join();
    }
    
    return 0;   
}

但我运行代码时，它可以打印数字，但会直接冻结，永远不会结束：

我要怎么做才能让它正常结束？

问题原因与解决方案 #

代码卡住是三个核心问题共同导致的，逐一修正即可：

• 条件变量的等待逻辑有漏洞：当前wait的谓词只判断队列非空，当生产者已经结束生产（finished=true）且队列已经空了的时候，还在等待的消费者会永远卡在wait调用上，不会被唤醒
• 共享变量访问没有加锁：queue、finished都是多线程共享的变量，你在生产者写数据、改finished，还有消费者while循环判断队列和结束标记的时候都没有加锁，存在数据竞争，属于C++标准规定的未定义行为
• 生产者设置完结束标记后没有唤醒消费者：生产者把finished设为true之后没有调用notify_all，还在等待的消费者收不到状态更新的通知

修正后的代码 #

#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>

using namespace std;

const int N_THREADS = 10;
const int N_TESTS = 100;
bool finished = false;
queue<int> q;
mutex m;
condition_variable cv;

void produce()
{
    for (int i = 0; i < N_TESTS; i++)
    {
        // 操作队列前先加锁
        unique_lock<mutex> lock(m);
        q.push(i);
        cv.notify_all();        
    }
    // 修改结束标记前加锁
    unique_lock<mutex> lock(m);
    finished = true;
    // 唤醒所有还在等待的消费者
    cv.notify_all();
}

void consume()
{
    while (true)
    {   
        unique_lock<mutex> lock(m);
        // 等待谓词同时判断队列非空或者生产已经结束
        cv.wait(lock, [] {return !q.empty() || finished; });
        // 队列空且生产结束，直接退出循环
        if (q.empty() && finished) {
            break;
        }
        int i = q.front();
        cout << i << endl;
        q.pop();
    }   
}

int main()
{
    thread producer(produce);
    
    vector<thread> consumers(N_THREADS);
    for (int i = 0; i < N_THREADS; i++)
    {
        consumers[i] = thread(consume);
    }

    producer.join();
    for (int i = 0; i < N_THREADS; i++)
    {
        consumers[i].join();
    }
    
    return 0;   
}

补充说明 #

核心逻辑是让消费者在生产结束且队列空的时候能正常退出，不要一直等待不会到来的新数据。另外所有对共享变量的读写操作都要放在锁的保护范围内，避免出现数据竞争导致的异常行为。

内容的提问来源于stack exchange，提问作者cheveuxdelin