使用unique_lock和condition_variable的多线程代码为何仅串行运行？

问题分析 #

你的代码存在几个关键问题导致任务只能串行执行：

1. 锁持有范围过大
   churn()函数中，unique_lock在获取后直到任务执行完成（包括sleep(1)）才释放，这意味着同一时刻只有一个线程能进入任务执行阶段，完全阻塞了并发。锁的作用应该仅局限于保护共享数据的访问，而非整个任务的执行流程。

2. 共享数据访问未加锁

   • do_work()中往wq队列添加任务时没有加锁，多线程同时操作队列会引发线程安全问题，导致队列状态混乱。
   • churn()开头的if (!wq.empty())检查在锁外进行，此时队列状态可能被其他线程修改，导致判断失效。

3. 线程创建逻辑错误
   每次调用do_work()就创建一个新线程，任务完成后线程销毁。这种方式不仅线程开销大，而且当线程进入churn()时，可能队列已经被其他线程取空，导致很多线程直接退出，无法形成稳定的并发执行池。

修复后的代码 #

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <map>
#include <thread>
#include <chrono>
#include <atomic>

class Work_Queue {
    std::mutex lock;
    std::condition_variable cv_task;
    std::condition_variable cv_result;
    int max_thrd;
    std::atomic<int> thrd_ctr; // 用atomic保证线程安全计数
    std::queue<std::pair<int, int>> wq;
    std::map<int, int> am;
    std::atomic<bool> stop; // 标记是否停止工作线程

public:
    Work_Queue(int cap) : max_thrd(cap), thrd_ctr(0), stop(false) {
        // 创建固定数量的工作线程
        for (int i = 0; i < max_thrd; ++i) {
            std::thread([this]() {
                while (!stop) {
                    std::unique_lock<std::mutex> ul(lock);
                    // 等待有任务或者停止信号
                    cv_task.wait(ul, [this]() { return !wq.empty() || stop; });
                    if (stop && wq.empty()) break;

                    auto [ip, id] = wq.front();
                    wq.pop();
                    ++thrd_ctr;
                    std::cout << "thrd # " << thrd_ctr << " started" << std::endl;
                    ul.unlock(); // 释放锁，让其他线程可以取任务

                    // 执行任务（无需锁保护）
                    int result = ip + 1;
                    std::this_thread::sleep_for(std::chrono::seconds(1));

                    ul.lock(); // 重新加锁更新共享数据
                    std::cout << "thrd # " << thrd_ctr << " ended" << std::endl;
                    am[id] = result;
                    --thrd_ctr;
                    cv_result.notify_all(); // 通知等待结果的线程
                }
            }).detach();
        }
    }

    ~Work_Queue() {
        stop = true;
        cv_task.notify_all(); // 唤醒所有工作线程退出
    }

    void do_work(int ip, int id) {
        std::lock_guard<std::mutex> lg(lock);
        wq.push({ip, id});
        cv_task.notify_one(); // 唤醒一个空闲的工作线程
    }

    int get_ans(int id) {
        std::unique_lock<std::mutex> ul(lock);
        cv_result.wait(ul, [this, id]() { return am.find(id) != am.end(); });
        return am[id];
    }
};

int main() {
    std::cout << "Number of cores: " << std::thread::hardware_concurrency() << std::endl;
    Work_Queue q(4);
    for (int i = 0; i < 6; ++i) {
        q.do_work(i, i + 5);
    }
    for (int i = 0; i < 6; ++i) {
        std::cout << q.get_ans(i + 5) << std::endl;
    }
    return 0;
}

修复说明 #

1. 调整锁范围：仅在访问共享队列wq、计数thrd_ctr和结果am时加锁，任务执行阶段释放锁，让多个线程可以同时执行任务。
2. 固定工作线程池：构造函数中创建指定数量的工作线程，这些线程循环等待任务，避免频繁创建销毁线程的开销。
3. 线程安全的计数与停止标志：用std::atomic保证计数和停止标志的线程安全访问。
4. 优化结果等待：get_ans改用条件变量等待结果，替代忙等待，减少CPU资源浪费。
5. 正确的通知机制：添加任务时唤醒一个工作线程，任务完成时唤醒等待结果的线程。

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