快捷導(dǎo)航

C++實(shí)現(xiàn)生產(chǎn)者與消費(fèi)者模式方式

更新時間：2025年12月28日 09:24:44 作者：star-keke

多線程工作池示例：創(chuàng)建固定數(shù)量的工作線程,通過條件變量競爭任務(wù)隊列,確保任務(wù)均勻分發(fā),任務(wù)隊列讀寫操作由互斥鎖保護(hù),避免競爭,使用`notify_one()`喚醒空閑線程,`notify_all()`停止時喚醒所有線程退出,適用于CPU/IO密集型任務(wù)

多線程工作池

創(chuàng)建workerCount個工作線程（示例中為 3 個），每個線程執(zhí)行相同的workerLoop邏輯。
線程通過condition_variable競爭任務(wù)隊列中的任務(wù)，確保任務(wù)被均勻分發(fā)。

線程安全保障

任務(wù)隊列的讀寫仍通過std::mutex保護(hù)，避免多線程競爭導(dǎo)致的數(shù)據(jù)錯亂。
cv.notify_one()每次喚醒一個線程處理任務(wù)，cv.notify_all()在停止時喚醒所有線程退出。

任務(wù)分發(fā)邏輯

生產(chǎn)者（主線程）提交任務(wù)時，通過notify_one()喚醒空閑線程，避免線程忙等。
多個工作線程同時消費(fèi)任務(wù)，提升任務(wù)處理效率（尤其適合 CPU/IO 密集型任務(wù)）。

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

// 任務(wù)隊列類型
using Task = std::function<void()>;

int main() {
    std::mutex mtx;
    std::condition_variable cv;
    std::queue<Task> taskQueue;
    bool stop = false; // 退出標(biāo)志
    const size_t workerCount = 3; // 工作線程數(shù)量
    std::vector<std::thread> workers; // 工作線程列表

    // ========== 工作線程循環(huán)：多線程消費(fèi)任務(wù) ==========
    auto workerLoop = [&](int workerId) {
        while (true) {
            Task task;

            // 加鎖，獲取任務(wù)或檢測退出信號
            {
                std::unique_lock<std::mutex> lock(mtx);
                
                // 等待條件：有任務(wù) 或 需要停止
                cv.wait(lock, [&]() {
                    return !taskQueue.empty() || stop;
                });

                // 若停止且任務(wù)隊列為空，退出循環(huán)
                if (stop && taskQueue.empty()) {
                    std::cout << "[線程" << workerId << "] 退出工作循環(huán)..." << std::endl;
                    break;
                }

                // 取出隊列頭部任務(wù)（多線程競爭，確保線程安全）
                task = std::move(taskQueue.front());
                taskQueue.pop();
                std::cout << "[線程" << workerId << "] 取出任務(wù)，準(zhǔn)備執(zhí)行..." << std::endl;
            } // 解鎖，避免執(zhí)行任務(wù)時持有鎖

            // 執(zhí)行任務(wù)
            if (task) {
                task();
            }
        }
    };

    // ========== 創(chuàng)建多個工作線程 ==========
    for (int i = 0; i < workerCount; ++i) {
        workers.emplace_back(workerLoop, i);
    }

    // ========== 模擬提交任務(wù)（生產(chǎn)者邏輯） ==========
    auto submitTask = [&](Task task) {
        std::lock_guard<std::mutex> lock(mtx);
        taskQueue.push(std::move(task));
        std::cout << "提交任務(wù)，當(dāng)前隊列大小：" << taskQueue.size() << std::endl;
        cv.notify_one(); // 喚醒一個等待的工作線程
    };

    // 批量提交10個任務(wù)
    for (int i = 0; i < 10; ++i) {
        submitTask([i]() {
            std::cout << "執(zhí)行任務(wù)" << i << "：線程ID=" << std::this_thread::get_id() << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(200)); // 模擬任務(wù)耗時
        });
    }

    // 等待所有任務(wù)執(zhí)行（可選，也可通過隊列狀態(tài)判斷）
    std::this_thread::sleep_for(std::chrono::seconds(3));

    // ========== 停止所有工作線程 ==========
    {
        std::lock_guard<std::mutex> lock(mtx);
        stop = true;
        cv.notify_all(); // 喚醒所有等待的線程，確保全部退出
        std::cout << "\n通知所有線程停止..." << std::endl;
    }

    // 等待所有工作線程結(jié)束
    for (auto& worker : workers) {
        if (worker.joinable()) {
            worker.join();
        }
    }

    std::cout << "程序結(jié)束" << std::endl;
    return 0;
}