C++多线程通信：构建高效的Master-Worker线程池模型

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional> // For std::function

// 定义一个任务类型，可以是任意可调用对象
using Task = std::function<void()>;

class ThreadPool {
public:
    ThreadPool(size_t num_threads) : stop(false) {
        for (size_t i = 0; i < num_threads; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    Task task;
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex);
                        // 等待任务到来或线程池停止
                        condition.wait(lock, [this] {
                            return !tasks.empty() || stop;
                        });

                        if (stop && tasks.empty()) {
                            return; // 线程池停止且无任务，退出
                        }
                        task = tasks.front();
                        tasks.pop();
                    }
                    task(); // 执行任务
                }
            });
        }
    }

    // 添加任务到队列
    void enqueue(Task task) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            tasks.push(std::move(task));
        }
        condition.notify_one(); // 通知一个等待中的工作线程
    }

    // 析构函数，停止所有工作线程
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all(); // 通知所有工作线程退出
        for (std::thread& worker : workers) {
            worker.join(); // 等待所有工作线程完成
        }
    }

private:
    std::vector<std::thread> workers;
    std::queue<Task> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

// 示例任务
void process_data(int id, const std::string& data) {
    std::cout << "Worker " << id << " processing: " << data << std::endl;
    // 模拟耗时操作
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

int main() {
    ThreadPool pool(4); // 创建一个包含4个工作线程的线程池

    // 模拟主线程分发任务
    for (int i = 0; i < 10; ++i) {
        std::string data = "Data_" + std::to_string(i);
        pool.enqueue([i, data]() { // 使用lambda捕获变量
            process_data(i, data);
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2)); // 等待任务执行
    // 线程池在main函数结束时自动析构并停止所有线程
    return 0;
}