C++ Master-Worker 模式：任务分发与结果聚合

好的，咱们开始吧！

大家好，我是你们今天的 C++ Master-Worker 模式导游。今天咱们要聊聊一个在并发编程里相当实用的设计模式：Master-Worker。这玩意儿就像一个高效的工厂流水线，老板（Master）负责分配任务，工人们（Workers）吭哧吭哧干活，最后老板再把结果汇总起来。听起来是不是很像你老板每天干的事情？

什么是 Master-Worker 模式？

简单来说，Master-Worker 模式是一种并行计算的设计模式。它将一个大的任务分解成若干个小的子任务，然后分配给多个 Worker 进程/线程并行执行。Master 进程/线程负责任务的分解、分配和结果的合并。

这模式能干嘛呢？举个例子：

图像处理： 将一张大图分割成小块，每个 Worker 处理一块，最后 Master 把处理后的图像拼接起来。
数据分析： 将海量数据分割成小份，每个 Worker 分析一份，Master 汇总分析结果。
网络爬虫： 将爬取任务分解成多个 URL，每个 Worker 负责爬取一部分 URL，Master 汇总爬取到的数据。
渲染： 将3D建模分割成多个场景，每个Worker渲染一部分，Master拼接渲染好的场景

总而言之，只要你能把一个大任务分解成可以并行执行的小任务，Master-Worker 模式就能帮你提高效率。

为什么要用 Master-Worker 模式？

提高效率： 并行执行，多核 CPU 充分利用起来，速度杠杠的。
可扩展性： Worker 数量可以动态调整，应对不同的任务规模。
容错性： 某个 Worker 挂了，不影响整个任务的执行，Master 可以重新分配任务。

Master-Worker 模式的组成部分

Master: 任务的分配者和结果的收集者。它负责：
- 将任务分解成子任务。
- 将子任务分配给 Worker。
- 接收 Worker 返回的结果。
- 合并结果。
- 管理 Worker 的生命周期（可选）。
Worker: 任务的执行者。它负责：
- 从 Master 接收任务。
- 执行任务。
- 将结果返回给 Master。
任务队列: 用于存储待执行的任务，Master 从队列中取出任务分配给 Worker。
结果队列: 用于存储 Worker 返回的结果，Master 从队列中取出结果进行合并。

一个简单的 C++ Master-Worker 示例

咱们先来一个最简单的例子，让你对 Master-Worker 模式有个直观的认识。这个例子里，Master 负责计算 1 到 100 的和，它把这个任务分解成 10 个子任务，每个 Worker 计算 10 个数的和。

#include <iostream>
#include <vector>
#include <thread>
#include <numeric>
#include <future>

// Worker 函数：计算一部分数的和
int worker(int start, int end) {
    int sum = 0;
    for (int i = start; i <= end; ++i) {
        sum += i;
    }
    return sum;
}

int main() {
    int num_workers = 10;
    int task_size = 10;
    std::vector<std::future<int>> results; // 用 future 来接收 worker 的结果
    std::vector<std::thread> workers;

    // 创建并启动 Worker 线程
    for (int i = 0; i < num_workers; ++i) {
        int start = i * task_size + 1;
        int end = (i + 1) * task_size;
        std::future<int> result = std::async(std::launch::async, worker, start, end); // 使用 async 启动线程并获取 future
        results.push_back(std::move(result));
    }

    // 等待所有 Worker 完成并收集结果
    int total_sum = 0;
    for (auto& result : results) {
        total_sum += result.get(); // get() 会阻塞直到结果可用
    }

    std::cout << "Total sum: " << total_sum << std::endl;

    return 0;
}

这个例子用到了 std::thread 创建线程，std::future 获取线程的返回值，std::async 简化了线程的创建和结果获取过程。

代码解释:

worker 函数: 这是 Worker 线程执行的函数，它计算 start 到 end 范围内的整数和。
main 函数:
- 定义了 Worker 的数量 num_workers 和每个 Worker 处理的任务大小 task_size。
- 创建了 std::vector<std::future<int>> results 来存储每个 Worker 返回的 future 对象。future 对象代表一个异步操作的结果。
- 循环创建并启动 Worker 线程。使用 std::async(std::launch::async, worker, start, end) 创建异步任务。std::launch::async 强制在新的线程中执行 worker 函数。std::async 返回一个 std::future<int> 对象，可以用来获取 worker 函数的返回值。
- 循环等待所有 Worker 完成并收集结果。使用 result.get() 获取 future 对象的结果。get() 函数会阻塞当前线程，直到结果可用。
- 将所有 Worker 返回的结果累加起来，得到最终的总和。

更复杂的 Master-Worker 示例：使用线程池和任务队列

上面的例子虽然简单，但缺少一些实际应用中需要的特性，比如线程池和任务队列。线程池可以避免频繁创建和销毁线程的开销，任务队列可以更好地管理待执行的任务。

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

// 线程池类
class ThreadPool {
public:
    ThreadPool(size_t num_threads) : num_threads_(num_threads), stop_(false) {
        threads_.reserve(num_threads_);
        for (size_t i = 0; i < num_threads_; ++i) {
            threads_.emplace_back([this] {
                while (true) {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(queue_mutex_);
                        condition_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
                        if (stop_ && tasks_.empty()) {
                            return;
                        }
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }

                    task();
                }
            });
        }
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (std::thread& thread : threads_) {
            thread.join();
        }
    }

    template<typename F>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            tasks_.emplace(std::forward<F>(f));
        }
        condition_.notify_one();
    }

private:
    std::vector<std::thread> threads_;
    std::queue<std::function<void()>> tasks_;
    std::mutex queue_mutex_;
    std::condition_variable condition_;
    size_t num_threads_;
    bool stop_;
};

// 任务类
struct Task {
    int id;
    int start;
    int end;
    std::promise<int> result; // 用 promise 来设置结果
};

// Master 类
class Master {
public:
    Master(size_t num_workers) : thread_pool_(num_workers) {}

    void submit_task(Task task) {
        thread_pool_.enqueue([this, task = std::move(task)]() mutable { // 捕获 task 的值拷贝
            int sum = 0;
            for (int i = task.start; i <= task.end; ++i) {
                sum += i;
            }
            task.result.set_value(sum); // 设置 promise 的值
        });
    }
private:
    ThreadPool thread_pool_;
};

int main() {
    int num_workers = 4;
    int num_tasks = 10;
    int task_size = 10;

    Master master(num_workers);
    std::vector<std::future<int>> results;

    for (int i = 0; i < num_tasks; ++i) {
        Task task;
        task.id = i;
        task.start = i * task_size + 1;
        task.end = (i + 1) * task_size;
        results.push_back(task.result.get_future()); // 获取 future 对象
        master.submit_task(std::move(task));
    }

    int total_sum = 0;
    for (auto& result : results) {
        total_sum += result.get();
    }

    std::cout << "Total sum: " << total_sum << std::endl;

    return 0;
}

代码解释:

ThreadPool 类:
- 构造函数创建指定数量的线程，每个线程都在一个循环中等待任务。
- enqueue 方法将任务添加到任务队列中，并通知一个等待的线程。
- 析构函数设置 stop_ 标志，通知所有线程退出循环，然后等待所有线程结束。
- 使用 std::mutex 和 std::condition_variable 来实现线程同步和任务队列的互斥访问。
Task 结构体: 包含任务的 ID、起始值、结束值，以及一个 std::promise<int> 对象。 promise 用于在任务执行完成后设置结果，对应的 future 对象可以用来获取结果。
Master 类:
- 构造函数创建一个线程池。
- submit_task 方法将任务提交到线程池中执行。它使用 lambda 表达式捕获 task 的值拷贝，避免多个线程同时访问同一个 task 对象。在 lambda 表达式中，计算任务的结果，然后使用 task.result.set_value(sum) 设置 promise 的值，从而将结果传递给主线程。
main 函数:
- 创建 Master 对象。
- 循环创建多个 Task 对象，并将它们提交给 Master。同时，将每个 Task 对象关联的 future 对象存储在 results 向量中。
- 循环等待所有任务完成，并累加结果。

关键点:

线程池： 避免频繁创建和销毁线程，提高效率。
任务队列： 管理待执行的任务，平衡 Master 和 Worker 的速度差异。
互斥锁和条件变量： 保证线程安全地访问任务队列。
std::promise 和 std::future： 在线程之间传递结果。 promise 允许一个线程设置一个值，而 future 允许另一个线程获取这个值。

Master-Worker 模式的优缺点

优点	缺点
提高效率，充分利用多核 CPU	实现复杂，需要考虑线程安全、任务分配策略、错误处理等问题。
可扩展性强，可以动态调整 Worker 数量	Master 节点容易成为瓶颈，需要考虑 Master 的负载均衡。
容错性好，某个 Worker 挂了不影响整体任务	需要考虑任务的分解粒度，太小会增加通信开销，太大不利于并行执行。
任务可以并行执行	对于某些不能分解的任务，Master-Worker 模式并不适用。

任务分解策略

任务分解是 Master-Worker 模式的关键。常见的任务分解策略有：

静态分解： 在任务开始前，将任务一次性分解成若干个子任务，然后分配给 Worker。适用于子任务大小相对均匀的情况。
动态分解： Master 根据 Worker 的执行情况，动态地分配任务。适用于子任务大小不均匀的情况，或者有 Worker 出现故障的情况。

任务分配策略

任务分配策略决定了 Master 如何将任务分配给 Worker。常见的任务分配策略有：

轮询： Master 按照顺序将任务分配给 Worker。简单易实现，但可能导致负载不均衡。
随机分配： Master 随机选择一个 Worker 分配任务。可以缓解轮询带来的负载不均衡问题。
负载均衡： Master 根据 Worker 的负载情况，选择负载较轻的 Worker 分配任务。需要收集 Worker 的负载信息，实现较为复杂。

错误处理

在 Master-Worker 模式中，错误处理非常重要。需要考虑以下几种情况：

Worker 崩溃： Master 需要检测到 Worker 崩溃，并将该 Worker 负责的任务重新分配给其他 Worker。
任务执行失败： Worker 执行任务失败，需要将错误信息返回给 Master，Master 可以选择重新执行该任务，或者放弃该任务。
Master 崩溃： Master 崩溃会导致整个任务失败，需要考虑 Master 的容错机制，比如使用备份 Master。

Master-Worker 模式的应用场景

大规模数据处理： 例如 MapReduce。
高性能计算： 例如科学计算、金融计算。
分布式系统： 例如 Web 服务器、缓存服务器。

总结

Master-Worker 模式是一种强大的并行计算设计模式，可以有效地提高任务的执行效率。但是，要正确地使用 Master-Worker 模式，需要仔细考虑任务分解、任务分配、错误处理等问题。希望今天的讲解能帮助你更好地理解和应用 Master-Worker 模式。

下次有机会，咱们再聊聊更高级的 Master-Worker 实现，比如使用消息队列、分布式锁等等。祝大家编程愉快！

发表回复 取消回复

发表回复取消回复