第30章案例实战

课程目录第30章案例实战

案例实战：并行计算框架——基于任务图的并行调度、动态负载均衡、性能基准测试

说实话，讲到这里，我觉得是时候把前面那些零散的知识点串起来了。你学了互斥锁、条件变量、线程池、无锁队列……但真正到了项目里，怎么把这些东西组合成一个能用的并行计算框架？

我个人习惯是，先画一张图，把整个框架的脉络理清楚。咱们今天要做的，就是一个基于任务图的并行调度系统。说白了，就是让一堆任务能自动找到依赖关系，然后并行执行，还能在运行时动态调整负载。

核心目标：构建一个支持任务依赖图、动态负载均衡、可基准测试的轻量级并行计算框架。

1. 任务图模型：先理清谁依赖谁

我在项目中遇到过最头疼的事，就是任务之间的依赖关系没理清，结果跑出来的结果全是错的。你想想看，任务B需要任务A的结果，结果B先跑了，那不就乱套了？

所以第一步，我们要定义任务图的数据结构。每个任务就是一个节点，依赖关系就是有向边。

// 任务节点
struct TaskNode {
    int id;
    std::string name;
    std::function<void()> work;          // 实际要执行的工作
    std::vector<int> dependencies;       // 依赖的任务ID列表
    std::atomic<int> remaining_deps;     // 还剩多少依赖没完成
    std::vector<int> dependents;         // 谁依赖我（反向边）
};

// 任务图
class TaskGraph {
public:
    void addTask(int id, std::string name, std::function<void()> work);
    void addDependency(int from, int to);  // from -> to
    void execute();  // 开始执行
private:
    std::unordered_map<int, std::unique_ptr<TaskNode>> nodes_;
    std::vector<int> ready_queue_;  // 初始就绪队列
};

嗯，这里要注意：remaining_deps 必须用原子变量。为什么？因为多个线程可能同时检查这个值，你不想看到数据竞争吧？

我的经验：任务ID最好用整数，别用字符串。字符串比较太慢了，尤其是在频繁查找依赖关系的时候。我曾经在一个项目里用字符串做ID，结果性能瓶颈全卡在字符串哈希上了。

2. 并行调度引擎：让任务跑起来

有了任务图，接下来就是调度引擎。核心逻辑其实很简单：

找到所有没有依赖的任务（入度为0），扔进就绪队列
线程池里的线程从就绪队列取任务执行
任务执行完后，通知所有依赖它的任务：我干完了
如果某个任务的依赖全部完成，把它加入就绪队列
重复直到所有任务完成

class TaskScheduler {
public:
    void execute(TaskGraph& graph) {
        // 1. 初始化：找出所有入度为0的任务
        auto& nodes = graph.getNodes();
        std::queue<int> ready;
        for (auto& [id, node] : nodes) {
            node->remaining_deps = node->dependencies.size();
            if (node->remaining_deps == 0) {
                ready.push(id);
            }
        }

        // 2. 用条件变量实现生产者-消费者模式
        std::mutex mtx;
        std::condition_variable cv;
        int completed = 0;
        int total = nodes.size();

        // 3. 启动线程池
        std::vector<std::thread> workers;
        for (int i = 0; i < num_threads_; ++i) {
            workers.emplace_back([&]() {
                while (true) {
                    int task_id;
                    {
                        std::unique_lock<std::mutex> lock(mtx);
                        cv.wait(lock, [&]() {
                            return !ready.empty() || completed == total;
                        });
                        if (completed == total) return;
                        task_id = ready.front();
                        ready.pop();
                    }

                    // 4. 执行任务
                    nodes[task_id]->work();

                    // 5. 通知依赖者
                    {
                        std::lock_guard<std::mutex> lock(mtx);
                        for (int dep_id : nodes[task_id]->dependents) {
                            if (--nodes[dep_id]->remaining_deps == 0) {
                                ready.push(dep_id);
                            }
                        }
                        ++completed;
                    }
                    cv.notify_all();
                }
            });
        }

        for (auto& w : workers) w.join();
    }
};

避坑指南：我曾经在通知依赖者的时候忘了加锁，结果两个线程同时修改同一个节点的 remaining_deps，导致计数错误。那个bug我调了整整一个下午。记住：任何共享数据的修改都需要同步。

3. 动态负载均衡：别让线程闲着

静态调度有个问题：如果某个任务特别重，其他线程都干完了，就等它一个，那并行度就浪费了。我建议用工作窃取（work stealing）来解决。

每个线程维护一个本地任务队列。当自己的队列空了，就去偷别人的任务。这样能自动平衡负载。

class WorkStealingQueue {
public:
    void push(int task_id) {
        std::lock_guard<std::mutex> lock(mtx_);
        queue_.push_back(task_id);
    }

    bool pop(int& task_id) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.empty()) return false;
        task_id = queue_.back();
        queue_.pop_back();
        return true;
    }

    bool steal(int& task_id) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.empty()) return false;
        task_id = queue_.front();  // 偷队头的任务
        queue_.pop_front();
        return true;
    }

private:
    std::deque<int> queue_;
    std::mutex mtx_;
};

你想想看，为什么偷任务要偷队头的，而自己取任务取队尾？这是为了减少冲突。每个线程主要操作自己的队尾，偷任务的操作队头，这样锁的竞争就小了。

4. 性能基准测试：用数据说话

框架写完了，怎么知道它快不快？我习惯用三组测试：

测试场景	任务数	依赖关系	预期效果
完全独立	1000	无依赖	接近线性加速
链式依赖	100	A→B→C→...	无并行度，退化为串行
混合依赖	500	随机DAG	部分并行，测试调度效率

// 基准测试代码
void benchmark() {
    TaskGraph graph;
    // 构造1000个独立任务
    for (int i = 0; i < 1000; ++i) {
        graph.addTask(i, "task_" + std::to_string(i), [i]() {
            // 模拟计算：计算斐波那契数列第30项
            volatile int result = fibonacci(30);
        });
    }

    TaskScheduler scheduler(4);  // 4线程
    auto start = std::chrono::high_resolution_clock::now();
    scheduler.execute(graph);
    auto end = std::chrono::high_resolution_clock::now();

    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
    std::cout << "耗时: " << duration.count() << " ms" << std::endl;
}

我的习惯：基准测试至少跑5次，取中位数。第一次跑的时候CPU缓存是冷的，结果会偏慢。另外，记得把测试程序设置成release模式，debug模式下的性能数据没有参考价值。

5. 整体架构图

下面这张图展示了整个框架的核心流程。我画的时候特意把数据流和控制流分开了，这样更容易理解。

6. 性能调优：那些年我踩过的坑

框架写好了，基准测试也跑了，但性能可能不理想。我总结几个常见问题：

任务粒度太细：如果每个任务只做几微秒的工作，线程切换的开销就超过了计算本身。我建议每个任务至少做1毫秒的工作。
锁竞争激烈：如果所有线程都在抢同一个锁，那就退化成串行了。试试用无锁队列或者分片锁。
缓存伪共享：多个线程修改相邻的内存地址，会导致缓存行失效。用 alignas(64) 把热点数据对齐到缓存行。

最终建议：不要过早优化。先把功能跑通，然后用性能分析工具（比如 perf、Intel VTune）找到真正的瓶颈，再针对性地优化。我见过太多人花了一周优化一个根本不存在的瓶颈。

好了，这个并行计算框架的核心思路就是这样。从任务图建模，到调度引擎，再到动态负载均衡，最后用基准测试验证性能。每一步都有坑，但每一步也都有对应的解决方案。你动手试试看，遇到问题再回来翻翻这部分内容。

📚 课程目录

案例实战：并行计算框架——基于任务图的并行调度、动态负载均衡、性能基准测试

1. 任务图模型：先理清谁依赖谁

2. 并行调度引擎：让任务跑起来

3. 动态负载均衡：别让线程闲着

4. 性能基准测试：用数据说话

5. 整体架构图

6. 性能调优：那些年我踩过的坑