前言你有没有想过每天凌晨3点系统是怎么自动执行数据同步、报表生成、缓存刷新这些定时任务的如果任务执行失败了系统会怎么处理分布式任务调度器是微服务架构中处理定时任务、异步任务的核心组件。今天我们用C语言从零实现一个分布式任务调度器的核心功能· 任务注册与发现· Cron表达式解析· 任务分片并行执行· 故障转移容错处理· 任务状态管理· 执行日志· 监控告警---一、调度器核心原理1. 架构图┌─────────────────────────────────────────────────────────────┐│ 调度器集群 ││ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ││ │ 节点1 │ │ 节点2 │ │ 节点3 │ ││ │ (Leader) │ │ (Follower) │ │ (Follower) │ ││ └──────┬──────┘ └─────────────┘ └─────────────┘ ││ │ ││ ▼ ││ ┌─────────────────────────────────────────────────────┐ ││ │ 任务分片策略 │ ││ │ [0-33%] [33%-66%] [66%-100%] │ ││ └─────────────────────────────────────────────────────┘ │└─────────────────────────────────────────────────────────────┘│ │ │▼ ▼ ▼┌─────────┐ ┌─────────┐ ┌─────────┐│ 执行器1 │ │ 执行器2 │ │ 执行器3 ││ 执行任务 │ │ 执行任务 │ │ 执行任务 │└─────────┘ └─────────┘ └─────────┘2. 核心概念概念 说明任务 需要定时执行的业务逻辑执行器 执行任务的工作节点分片 大任务拆分成小片并行执行故障转移 节点宕机任务转移给其他节点Cron 定时表达式---二、完整代码实现1. 基础数据结构c#include stdio.h#include stdlib.h#include string.h#include unistd.h#include pthread.h#include time.h#include errno.h#include sys/socket.h#include netinet/in.h#include arpa/inet.h#define MAX_TASK_NAME 128#define MAX_EXECUTOR 64#define MAX_SHARDS 100#define MAX_LOG_LEN 1024// 任务状态typedef enum {TASK_IDLE 0,TASK_RUNNING,TASK_SUCCESS,TASK_FAILED,TASK_TIMEOUT,TASK_CANCELLED} task_status_t;// 任务分片typedef struct task_shard {int shard_index;int total_shards;char executor[MAX_EXECUTOR];task_status_t status;time_t start_time;time_t end_time;char error_msg[256];struct task_shard *next;} task_shard_t;// 任务定义typedef struct task_definition {char name[MAX_TASK_NAME];char cron_expr[64];int shard_count;int timeout_seconds;int retry_count;void (*execute)(task_shard_t *shard);struct task_definition *next;} task_definition_t;// 任务实例typedef struct task_instance {char task_name[MAX_TASK_NAME];char instance_id[64];task_shard_t *shards;int shard_count;int completed_shards;task_status_t status;time_t start_time;time_t end_time;struct task_instance *next;} task_instance_t;// 执行器节点typedef struct executor_node {char name[MAX_EXECUTOR];char host[32];int port;int healthy;time_t last_heartbeat;struct executor_node *next;} executor_node_t;// 任务日志typedef struct task_log {char task_name[MAX_TASK_NAME];char instance_id[64];int shard_index;time_t log_time;char level[16];char message[MAX_LOG_LEN];struct task_log *next;} task_log_t;// 任务调度器typedef struct task_scheduler {task_definition_t *tasks;task_instance_t *instances;executor_node_t *executors;task_log_t *logs;int executor_count;int running;pthread_mutex_t mutex;pthread_t schedule_thread;pthread_t health_check_thread;int port;} task_scheduler_t;2. 任务注册c// 创建调度器task_scheduler_t *scheduler_create(int port) {task_scheduler_t *s malloc(sizeof(task_scheduler_t));memset(s, 0, sizeof(task_scheduler_t));s-port port;s-running 1;pthread_mutex_init(s-mutex, NULL);printf([调度器] 启动端口: %d\n, port);return s;}// 注册任务void scheduler_register_task(task_scheduler_t *s, const char *name,const char *cron_expr, int shard_count,int timeout_seconds, int retry_count,void (*execute)(task_shard_t*)) {pthread_mutex_lock(s-mutex);task_definition_t *task malloc(sizeof(task_definition_t));strcpy(task-name, name);strcpy(task-cron_expr, cron_expr);task-shard_count shard_count 0 ? shard_count : 1;task-timeout_seconds timeout_seconds 0 ? timeout_seconds : 300;task-retry_count retry_count 0 ? retry_count : 3;task-execute execute;task-next s-tasks;s-tasks task;pthread_mutex_unlock(s-mutex);printf([任务] 注册: %s, cron%s, shards%d\n, name, cron_expr, shard_count);}// 注册执行器void scheduler_register_executor(task_scheduler_t *s, const char *name,const char *host, int port) {pthread_mutex_lock(s-mutex);executor_node_t *node malloc(sizeof(executor_node_t));strcpy(node-name, name);strcpy(node-host, host);node-port port;node-healthy 1;node-last_heartbeat time(NULL);node-next s-executors;s-executors node;s-executor_count;pthread_mutex_unlock(s-mutex);printf([执行器] 注册: %s (%s:%d)\n, name, host, port);}3. Cron解析c// Cron字段typedef struct cron_fields {int minute[60];int minute_count;int hour[24];int hour_count;int day[31];int day_count;int month[12];int month_count;int weekday[7];int weekday_count;} cron_fields_t;// 解析Cron表达式int cron_parse(const char *expr, cron_fields_t *fields) {char parts[6][32];int count sscanf(expr, %s %s %s %s %s,parts[0], parts[1], parts[2], parts[3], parts[4]);if (count ! 5) return -1;// 解析分钟if (strcmp(parts[0], *) 0) {fields-minute_count 60;for (int i 0; i 60; i) fields-minute[i] i;} else {fields-minute_count 1;fields-minute[0] atoi(parts[0]);}// 解析小时if (strcmp(parts[1], *) 0) {fields-hour_count 24;for (int i 0; i 24; i) fields-hour[i] i;} else {fields-hour_count 1;fields-hour[0] atoi(parts[1]);}// 解析日if (strcmp(parts[2], *) 0) {fields-day_count 31;for (int i 1; i 31; i) fields-day[i-1] i;} else {fields-day_count 1;fields-day[0] atoi(parts[2]);}// 解析月if (strcmp(parts[3], *) 0) {fields-month_count 12;for (int i 1; i 12; i) fields-month[i-1] i;} else {fields-month_count 1;fields-month[0] atoi(parts[3]);}// 解析周if (strcmp(parts[4], *) 0) {fields-weekday_count 7;for (int i 0; i 7; i) fields-weekday[i] i;} else {fields-weekday_count 1;fields-weekday[0] atoi(parts[4]);}return 0;}int cron_match(cron_fields_t *fields, struct tm *tm) {int match 0;for (int i 0; i fields-minute_count; i) {if (fields-minute[i] tm-tm_min) { match 1; break; }}if (!match) return 0;match 0;for (int i 0; i fields-hour_count; i) {if (fields-hour[i] tm-tm_hour) { match 1; break; }}if (!match) return 0;match 0;for (int i 0; i fields-day_count; i) {if (fields-day[i] tm-tm_mday) { match 1; break; }}if (!match) return 0;match 0;for (int i 0; i fields-month_count; i) {if (fields-month[i] tm-tm_mon 1) { match 1; break; }}if (!match) return 0;match 0;for (int i 0; i fields-weekday_count; i) {if (fields-weekday[i] tm-tm_wday) { match 1; break; }}return match;}4. 任务调度c// 执行分片任务void execute_shard(task_scheduler_t *s, task_definition_t *task,task_instance_t *instance, int shard_idx) {task_shard_t *shard instance-shards[shard_idx];shard-shard_index shard_idx;shard-total_shards task-shard_count;strcpy(shard-executor, local);shard-status TASK_RUNNING;shard-start_time time(NULL);printf([执行] 分片 %d/%d: %s\n, shard_idx1, task-shard_count, task-name);if (task-execute) {task-execute(shard);shard-status TASK_SUCCESS;}shard-end_time time(NULL);pthread_mutex_lock(s-mutex);instance-completed_shards;if (instance-completed_shards task-shard_count) {instance-status TASK_SUCCESS;instance-end_time time(NULL);}pthread_mutex_unlock(s-mutex);}// 检查并触发任务void scheduler_check_tasks(task_scheduler_t *s) {time_t now time(NULL);struct tm *tm_now localtime(now);pthread_mutex_lock(s-mutex);task_definition_t *task s-tasks;while (task) {cron_fields_t fields;if (cron_parse(task-cron_expr, fields) 0) {if (cron_match(fields, tm_now)) {// 检查是否有实例在运行task_instance_t *inst s-instances;int running 0;while (inst) {if (strcmp(inst-task_name, task-name) 0 inst-status TASK_RUNNING) {running 1;break;}inst inst-next;}if (!running) {// 创建新任务实例task_instance_t *new_inst malloc(sizeof(task_instance_t));strcpy(new_inst-task_name, task-name);snprintf(new_inst-instance_id, sizeof(new_inst-instance_id),%s-%ld, task-name, now);new_inst-shard_count task-shard_count;new_inst-shards calloc(task-shard_count, sizeof(task_shard_t));new_inst-completed_shards 0;new_inst-status TASK_RUNNING;new_inst-start_time now;new_inst-end_time 0;new_inst-next s-instances;s-instances new_inst;printf([调度] 触发任务: %s\n, task-name);// 执行所有分片for (int i 0; i task-shard_count; i) {execute_shard(s, task, new_inst, i);}}}}task task-next;}pthread_mutex_unlock(s-mutex);}5. 故障转移c// 检查执行器健康void scheduler_check_health(task_scheduler_t *s) {pthread_mutex_lock(s-mutex);time_t now time(NULL);executor_node_t *node s-executors;while (node) {if (now - node-last_heartbeat 30) {if (node-healthy) {node-healthy 0;printf([告警] 执行器 %s 失联\n, node-name);}}node node-next;}pthread_mutex_unlock(s-mutex);}// 故障转移void scheduler_failover(task_scheduler_t *s) {pthread_mutex_lock(s-mutex);time_t now time(NULL);task_instance_t *inst s-instances;while (inst) {if (inst-status TASK_RUNNING) {for (int i 0; i inst-shard_count; i) {if (inst-shards[i].status TASK_RUNNING) {time_t elapsed now - inst-shards[i].start_time;if (elapsed 300) {inst-shards[i].status TASK_FAILED;strcpy(inst-shards[i].error_msg, Timeout, failover);printf([故障转移] 分片 %d 超时\n, i);}}}}inst inst-next;}pthread_mutex_unlock(s-mutex);}6. 测试代码c// 示例任务void data_sync_task(task_shard_t *shard) {printf([任务] 数据同步 分片 %d/%d\n, shard-shard_index1, shard-total_shards);sleep(1);}void report_task(task_shard_t *shard) {printf([任务] 报表生成 分片 %d/%d\n, shard-shard_index1, shard-total_shards);sleep(2);}void test_scheduler() {printf( 分布式任务调度器测试 \n\n);task_scheduler_t *s scheduler_create(8080);// 注册任务scheduler_register_task(s, data-sync, 0 0 3 * * *, 4, 300, 3, data_sync_task);scheduler_register_task(s, report-gen, 0 30 2 * * *, 2, 600, 2, report_task);// 注册执行器scheduler_register_executor(s, executor-1, 127.0.0.1, 9001);scheduler_register_executor(s, executor-2, 127.0.0.1, 9002);// 模拟调度手动触发printf(\n模拟调度任务...\n);scheduler_check_tasks(s);printf(\n任务实例:\n);task_instance_t *inst s-instances;while (inst) {printf( %s: 状态%d, 完成%d/%d\n,inst-task_name, inst-status,inst-completed_shards, inst-shard_count);inst inst-next;}free(s);}int main() {test_scheduler();return 0;}---三、编译和运行bashgcc -o task_scheduler task_scheduler.c -lpthread./task_scheduler---四、XXL-JOB vs 本实现特性 本实现 XXL-JOB任务注册 ✅ ✅Cron调度 ✅ ✅任务分片 ✅ ✅故障转移 ✅ ✅执行器管理 ✅ ✅可视化界面 ❌ ✅任务依赖 ❌ ❌任务编排 ❌ ❌---五、总结通过这篇文章你学会了· 分布式调度器的核心原理· Cron表达式解析· 任务分片与并行执行· 故障转移机制· 执行器健康检查· 任务状态管理分布式调度器是大数据平台和微服务架构的核心组件。掌握它你就理解了数据同步、报表生成、定时清理等系统的设计原理。下一篇预告《从零实现一个分布式ID生成器雪花算法与号段模式》---评论区分享一下你用调度器解决过什么场景
