实战项目三：分布式任务调度系统
好，终于到了第三个实战项目。说实话，前面两个项目——消息队列和事件驱动框架——都是在为这个项目打基础。分布式任务调度系统，说白了就是把一堆任务拆开，扔到不同的机器上去跑，然后统一收结果。
我在做物联网平台的时候，就遇到过类似的需求。当时有几十万台设备定时上报数据，单机根本扛不住。后来我设计了一套分布式调度系统，把任务按区域分片，丢到不同的计算节点上去处理。嗯，今天这个项目，就是那个系统的简化版。
项目背景与需求
假设你正在开发一个视频转码平台。用户上传一个视频，系统需要把它转成多种分辨率（720p、1080p、4K）。单机转码太慢，用户等得不耐烦。你需要把转码任务拆开，分发给多台机器并行处理。
核心需求其实就三条：
- 任务分发：把一个大任务拆成多个子任务，发给不同的工作节点
- 状态跟踪：随时知道每个子任务跑到哪一步了
- 失败重试：某个节点挂了，任务不能丢，得重新调度
架构核心思想：用消息队列做任务缓冲，用事件驱动做状态流转。说白了就是——任务来了先排队，处理完了发个事件通知，下一个环节自动触发。
系统整体架构
先画一张图，让你对整个系统有个直观印象。我个人习惯在动手写代码之前，先把架构图画清楚。这样后面编码的时候，思路不会乱。
你看这张图，其实不复杂。用户提交任务给调度器，调度器把任务拆成子任务，扔进消息队列。Worker 节点从队列里抢任务，处理完了把结果回传给调度器。调度器收到结果后，更新任务状态，决定下一步做什么。
核心数据结构设计
先定义任务相关的数据结构。我在项目中吃过亏——一开始没想清楚状态机，后面改得想哭。所以这次咱们先把状态定义清楚。
// 任务状态枚举
typedef enum {
TASK_PENDING = 0, // 等待调度
TASK_RUNNING = 1, // 正在执行
TASK_COMPLETED = 2, // 执行成功
TASK_FAILED = 3, // 执行失败
TASK_TIMEOUT = 4 // 超时
} TaskState;
// 子任务结构体
typedef struct {
uint32_t task_id; // 任务ID
uint32_t subtask_id; // 子任务ID
TaskState state; // 当前状态
char payload[256]; // 任务数据（比如视频片段路径）
uint32_t retry_count; // 已重试次数
uint32_t max_retries; // 最大重试次数
uint64_t timeout_ms; // 超时时间
} SubTask;
// 任务调度上下文
typedef struct {
uint32_t total_subtasks; // 子任务总数
uint32_t completed_count; // 已完成数
uint32_t failed_count; // 失败数
SubTask subtasks[32]; // 子任务列表
void (*on_complete)(void*); // 全部完成回调
} TaskContext;个人经验：retry_count 和 max_retries 一定要分开。我曾经图省事，只留了一个 retry_count，结果重试逻辑写成了死循环。嗯，血的教训。
消息队列封装
消息队列是系统的核心纽带。我们基于之前课程写的环形缓冲区，封装一个支持多生产者多消费者的队列。
// 消息队列句柄
typedef struct {
RingBuffer* buffer; // 底层环形缓冲区
SemaphoreHandle_t mutex; // 互斥锁
SemaphoreHandle_t items_avail; // 有数据可读的信号量
SemaphoreHandle_t slots_avail; // 有空闲槽位的信号量
} MsgQueue;
// 初始化消息队列
int msg_queue_init(MsgQueue* mq, uint32_t capacity) {
mq->buffer = ring_buffer_create(capacity, sizeof(SubTask));
mq->mutex = xSemaphoreCreateMutex();
mq->items_avail = xSemaphoreCreateCounting(capacity, 0);
mq->slots_avail = xSemaphoreCreateCounting(capacity, capacity);
return (mq->buffer && mq->mutex && mq->items_avail && mq->slots_avail) ? 0 : -1;
}
// 发送任务（生产者调用）
int msg_queue_send(MsgQueue* mq, SubTask* task) {
// 等待有空闲槽位
if (xSemaphoreTake(mq->slots_avail, pdMS_TO_TICKS(1000)) != pdTRUE) {
return -1; // 队列满了，超时
}
xSemaphoreTake(mq->mutex, portMAX_DELAY);
ring_buffer_write(mq->buffer, task);
xSemaphoreGive(mq->mutex);
xSemaphoreGive(mq->items_avail); // 通知消费者
return 0;
}
// 接收任务（消费者调用）
int msg_queue_recv(MsgQueue* mq, SubTask* task, uint32_t timeout_ms) {
if (xSemaphoreTake(mq->items_avail, pdMS_TO_TICKS(timeout_ms)) != pdTRUE) {
return -1; // 没有任务，超时
}
xSemaphoreTake(mq->mutex, portMAX_DELAY);
ring_buffer_read(mq->buffer, task);
xSemaphoreGive(mq->mutex);
xSemaphoreGive(mq->slots_avail); // 释放槽位
return 0;
}你想想看，为什么用两个信号量？一个管「有没有数据」，一个管「有没有空位」。这样生产者不会覆盖未读的数据，消费者不会读到空数据。说白了就是经典的「生产者-消费者」模型。
任务调度器实现
调度器是整个系统的大脑。它的职责是：接收任务、拆分子任务、监控进度、处理失败。
// 调度器主结构
typedef struct {
MsgQueue task_queue; // 任务队列
TaskContext active_tasks[16]; // 正在进行的任务列表
uint32_t task_count; // 活跃任务数
void (*status_callback)(uint32_t task_id, TaskState state);
} Scheduler;
// 提交一个转码任务
int scheduler_submit(Scheduler* sched, uint32_t task_id,
const char* video_path, uint32_t num_workers) {
// 拆分子任务：每个分辨率一个子任务
const char* resolutions[] = {"720p", "1080p", "4K"};
uint32_t num_res = sizeof(resolutions) / sizeof(resolutions[0]);
TaskContext* ctx = &sched->active_tasks[sched->task_count++];
ctx->total_subtasks = num_res;
ctx->completed_count = 0;
ctx->failed_count = 0;
for (int i = 0; i < num_res; i++) {
SubTask st;
st.task_id = task_id;
st.subtask_id = i;
st.state = TASK_PENDING;
st.retry_count = 0;
st.max_retries = 3;
st.timeout_ms = 30000; // 30秒超时
// 打包任务数据
snprintf(st.payload, sizeof(st.payload),
"%s|%s", video_path, resolutions[i]);
// 扔进队列
msg_queue_send(&sched->task_queue, &st);
}
return 0;
}
// 处理子任务完成事件
void scheduler_on_subtask_done(Scheduler* sched,
uint32_t task_id,
uint32_t subtask_id,
int success) {
// 找到对应的任务上下文
TaskContext* ctx = NULL;
for (int i = 0; i < sched->task_count; i++) {
if (sched->active_tasks[i].subtasks[0].task_id == task_id) {
ctx = &sched->active_tasks[i];
break;
}
}
if (!ctx) return;
if (success) {
ctx->completed_count++;
ctx->subtasks[subtask_id].state = TASK_COMPLETED;
} else {
ctx->failed_count++;
ctx->subtasks[subtask_id].state = TASK_FAILED;
// 检查是否需要重试
if (ctx->subtasks[subtask_id].retry_count <
ctx->subtasks[subtask_id].max_retries) {
ctx->subtasks[subtask_id].retry_count++;
ctx->subtasks[subtask_id].state = TASK_PENDING;
msg_queue_send(&sched->task_queue,
&ctx->subtasks[subtask_id]);
}
}
// 检查是否全部完成
if (ctx->completed_count + ctx->failed_count >= ctx->total_subtasks) {
if (ctx->on_complete) {
ctx->on_complete(&task_id);
}
}
}避坑指南：我曾经在重试逻辑里忘记重置 state 为 TASK_PENDING，结果任务重试时状态还是 TASK_FAILED，Worker 节点直接跳过了。嗯，这种低级错误，犯一次就够了。
Worker 节点实现
Worker 节点相对简单——从队列拿任务，执行，回传结果。但要注意，Worker 不能假设任务永远成功。
// Worker 节点主循环
void worker_run(MsgQueue* queue, uint32_t worker_id) {
SubTask task;
while (1) {
// 从队列取任务，超时设为 5000ms
int ret = msg_queue_recv(queue, &task, 5000);
if (ret != 0) {
// 没任务，休眠一会儿再试
vTaskDelay(pdMS_TO_TICKS(100));
continue;
}
// 更新状态为运行中
task.state = TASK_RUNNING;
printf("[Worker %d] 开始处理任务 %d-%d\n",
worker_id, task.task_id, task.subtask_id);
// 执行实际任务（这里是模拟转码）
int success = execute_transcode_task(task.payload);
// 回传结果
if (success) {
task.state = TASK_COMPLETED;
printf("[Worker %d] 任务 %d-%d 完成\n",
worker_id, task.task_id, task.subtask_id);
} else {
task.state = TASK_FAILED;
printf("[Worker %d] 任务 %d-%d 失败\n",
worker_id, task.task_id, task.subtask_id);
}
// 通过事件通知调度器
event_publish(EVENT_SUBTASK_DONE, &task, sizeof(SubTask));
}
}
// 模拟转码任务
static int execute_transcode_task(const char* payload) {
// 解析 payload: "video_path|resolution"
char path[128], res[16];
sscanf(payload, "%[^|]|%s", path, res);
// 模拟处理时间
vTaskDelay(pdMS_TO_TICKS(rand() % 3000 + 1000));
// 模拟 10% 的失败率
return (rand() % 10) > 0;
}事件驱动状态流转
整个系统的状态流转，全靠事件驱动。我画了一张状态图，你一看就明白。
状态流转的逻辑其实很清晰：PENDING 是起点，调度器把任务丢进队列后，状态就是 PENDING。Worker 取走任务，变成 RUNNING。执行成功变 COMPLETED，失败变 FAILED。FAILED 之后，如果重试次数没超限，重新变回 PENDING 再排一次队。
完整运行示例
最后，咱们把整个系统串起来跑一遍。我习惯写一个简单的 main 函数来验证整个流程。
// 事件回调：处理子任务完成事件
void on_subtask_done(void* arg, void* data, uint32_t len) {
SubTask* task = (SubTask*)data;
scheduler_on_subtask_done(&g_scheduler,
task->task_id,
task->subtask_id,
task->state == TASK_COMPLETED);
}
// 任务全部完成的回调
void on_all_complete(void* arg) {
uint32_t* task_id = (uint32_t*)arg;
printf("任务 %d 全部子任务处理完成！\n", *task_id);
}
int main() {
Scheduler sched;
MsgQueue queue;
// 初始化
scheduler_init(&sched, &queue);
event_register(EVENT_SUBTASK_DONE, on_subtask_done, NULL);
// 启动 3 个 Worker 节点
for (int i = 0; i < 3; i++) {
xTaskCreate(worker_task, "worker", 2048, &queue, 1, NULL);
}
// 提交一个转码任务
scheduler_submit(&sched, 1001, "/videos/demo.mp4", 3);
// 启动调度器主循环
scheduler_run(&sched);
return 0;
}核心要点回顾：
- 消息队列做解耦：生产者和消费者不直接通信，通过队列中转
- 事件驱动做联动：子任务完成自动触发状态更新，不需要轮询
- 状态机做容错：失败自动重试，重试超限才放弃
这个项目做完，你基本上就掌握了分布式任务调度的核心套路。说白了就是三件事：拆任务、派任务、收结果。消息队列和事件驱动，就是帮你把这三件事串起来的骨架。
我在实际项目中，这套架构支撑过每天百万级的任务调度。当然，生产环境要考虑的东西更多——持久化、负载均衡、服务发现等等。但核心思想不变，你先把今天这个跑通了，后面加功能就是水到渠成的事。
扩展思考：如果 Worker 节点本身也挂了怎么办？嗯，可以加一个心跳机制——Worker 定期向调度器报活，超时没报活就把它的任务重新分配给其他 Worker。这个留给你自己实现试试。
公众号:蓝海资料掘金营，微信deep3321