在C++项目开发中，经常会遇到多个异步任务存在先后依赖的场景，比如先异步拉取配置，再基于配置异步加载数据，最后异步处理结果。传统的回调嵌套方式会导致代码逻辑分散、可读性差，维护成本很高。本文介绍一种任务链式异步执行模型的设计方案，通过封装任务节点、管理依赖关系，实现异步任务的按顺序链式执行，同时提供完整的实战案例源码。
模型核心设计思路
该模型的核心是将每个异步任务封装为独立的任务节点，节点包含任务执行逻辑、依赖的前置任务、执行完成后的回调。所有任务节点注册到任务调度器中，调度器监听前置任务的完成状态，当所有前置任务完成后自动触发当前任务的执行，形成链式执行效果。
核心组件说明
- TaskNode：任务节点类，封装单个异步任务的执行逻辑、前置依赖、完成状态
- TaskChain：任务链管理类，负责任务节点的注册、依赖关系校验、调度执行
- TaskScheduler：任务调度器，基于线程池执行异步任务，触发任务完成后的后续流程
关键实现代码
任务节点封装实现
首先实现任务节点的基础封装，使用std::promise和std::future管理任务执行结果和状态：
#include <iostream>
#include <future>
#include <functional>
#include <vector>
#include <memory>
#include <mutex>
#include <thread>
#include <queue>
// 任务节点基类
class TaskNode : public std::enable_shared_from_this<TaskNode> {
public:
using TaskFunc = std::function<void(std::shared_ptr<TaskNode>)>;
using CompleteCallback = std::function<void(std::shared_ptr<TaskNode>)>;
TaskNode(TaskFunc func) : task_func_(func), is_completed_(false) {}
// 执行任务
void Run() {
if (task_func_) {
task_func_(shared_from_this());
}
}
// 标记任务完成
void SetCompleted() {
std::lock_guard<std::mutex> lock(mutex_);
is_completed_ = true;
promise_.set_value();
// 触发完成回调
if (complete_callback_) {
complete_callback_(shared_from_this());
}
}
// 获取任务完成的future
std::future<void> GetFuture() {
return promise_.get_future();
}
// 设置完成回调
void SetCompleteCallback(CompleteCallback callback) {
complete_callback_ = callback;
}
// 添加前置依赖任务
void AddDependency(std::shared_ptr<TaskNode> dep) {
dependencies_.push_back(dep);
}
// 检查所有前置依赖是否完成
bool AreDependenciesCompleted() {
for (auto& dep : dependencies_) {
if (!dep->is_completed_) {
return false;
}
}
return true;
}
private:
TaskFunc task_func_;
CompleteCallback complete_callback_;
std::promise<void> promise_;
std::vector<std::shared_ptr<TaskNode>> dependencies_;
bool is_completed_;
std::mutex mutex_;
};
任务链与调度器实现
接下来实现任务调度器和任务链管理逻辑，负责按依赖关系调度任务执行：
// 简单线程池实现
class ThreadPool {
public:
ThreadPool(size_t thread_num) {
for (size_t i = 0; i < thread_num; ++i) {
workers_.emplace_back([this]() {
while (true) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(queue_mutex_);
condition_.wait(lock, [this]() { return !tasks_.empty() || stop_; });
if (stop_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
});
}
}
template<class F>
void Enqueue(F&& f) {
{
std::lock_guard<std::mutex> lock(queue_mutex_);
tasks_.emplace(std::forward<F>(f));
}
condition_.notify_one();
}
~ThreadPool() {
{
std::lock_guard<std::mutex> lock(queue_mutex_);
stop_ = true;
}
condition_.notify_all();
for (auto& worker : workers_) {
if (worker.joinable()) worker.join();
}
}
private:
std::vector<std::thread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex queue_mutex_;
std::condition_variable condition_;
bool stop_ = false;
};
// 任务链调度器
class TaskChainScheduler {
public:
TaskChainScheduler(size_t thread_num = 4) : thread_pool_(thread_num) {}
// 注册任务到链中
void RegisterTask(std::shared_ptr<TaskNode> task) {
std::lock_guard<std::mutex> lock(tasks_mutex_);
tasks_.push_back(task);
// 设置任务完成回调，检查后续可执行的任务
task->SetCompleteCallback([this](std::shared_ptr<TaskNode> completed_task) {
CheckAndRunNextTasks();
});
}
// 启动任务链执行
void Start() {
CheckAndRunNextTasks();
}
private:
// 检查并执行所有可执行的任务
void CheckAndRunNextTasks() {
std::lock_guard<std::mutex> lock(tasks_mutex_);
for (auto& task : tasks_) {
if (!task->AreDependenciesCompleted()) continue;
// 检查任务是否已经执行
auto future = task->GetFuture();
if (future.wait_for(std::chrono::seconds(0)) == std::future_status::ready) continue;
// 提交任务到线程池执行
thread_pool_.Enqueue([task]() {
task->Run();
});
}
}
ThreadPool thread_pool_;
std::vector<std::shared_ptr<TaskNode>> tasks_;
std::mutex tasks_mutex_;
};
实战案例演示
下面通过一个三个异步任务链式执行的案例，演示模型的使用方式：
int main() {
// 创建调度器
TaskChainScheduler scheduler;
// 任务1：模拟异步拉取配置
auto task1 = std::make_shared<TaskNode>([](std::shared_ptr<TaskNode> self) {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "任务1：拉取配置完成" << std::endl;
self->SetCompleted();
});
// 任务2：依赖任务1，模拟异步加载数据
auto task2 = std::make_shared<TaskNode>([](std::shared_ptr<TaskNode> self) {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "任务2：基于配置加载数据完成" << std::endl;
self->SetCompleted();
});
task2->AddDependency(task1);
// 任务3：依赖任务2，模拟异步处理结果
auto task3 = std::make_shared<TaskNode>([](std::shared_ptr<TaskNode> self) {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "任务3：处理数据结果完成" << std::endl;
self->SetCompleted();
});
task3->AddDependency(task2);
// 注册所有任务
scheduler.RegisterTask(task1);
scheduler.RegisterTask(task2);
scheduler.RegisterTask(task3);
// 启动任务链
scheduler.Start();
// 等待所有任务完成
task3->GetFuture().wait();
std::cout << "所有异步任务链式执行完成" << std::endl;
return 0;
}
模型优化方向
上述基础实现可以满足简单的链式异步执行需求，实际使用中还可以做以下优化：
- 增加任务超时机制，避免某个任务异常阻塞整个链条
- 支持任务执行失败的重试逻辑，提升模型健壮性
- 增加任务优先级配置，支持不同优先级的任务调度
- 提供任务执行结果的传递机制，让后续任务可以获取前置任务的返回数据
该模型通过封装任务节点和依赖关系，避免了回调嵌套的问题，让异步任务的逻辑更清晰，扩展起来也更方便，适合在需要处理多异步依赖的C++项目中应用。
C++异步任务任务链式执行异步执行模型std::futurestd::promise修改时间：2026-06-09 15:24:40