前言上篇文章我们实现了Airflow的基础功能今天我们来深入实现更多进阶特性让任务调度更接近生产级· XCom任务间数据传递· 任务重试与回填Backfill· 触发规则Trigger Rules· 传感器Sensor——等待外部条件· 工作池Pool——资源隔离· 连接管理Connection· 任务超时与死锁检测· 历史执行追踪---一、进阶概念概念 说明XCom 任务间共享数据Backfill 回填历史数据Trigger Rule 触发条件ALL_SUCCESS/ALL_FAILED/ONE_SUCCESSSensor 等待外部条件满足Pool 资源隔离池Connection 外部系统连接管理---二、完整代码实现1. 数据传递XComc#include stdio.h#include stdlib.h#include string.h#include unistd.h#include pthread.h#include time.h#include errno.h#define MAX_XCOM_KEY 64#define MAX_XCOM_VALUE 1024#define MAX_TASK_NAME 64#define MAX_RETRIES 5// XCom条目typedef struct xcom_entry {char key[MAX_XCOM_KEY];char value[MAX_XCOM_VALUE];char task_id[MAX_TASK_NAME];char dag_id[64];time_t timestamp;struct xcom_entry *next;} xcom_entry_t;// 任务状态typedef enum {TASK_NONE 0,TASK_SCHEDULED,TASK_RUNNING,TASK_SUCCESS,TASK_FAILED,TASK_UPSTREAM_FAILED,TASK_SKIPPED,TASK_UP_FOR_RETRY} task_state_t;// 触发规则typedef enum {TRIGGER_ALL_SUCCESS 0,TRIGGER_ALL_FAILED,TRIGGER_ALL_DONE,TRIGGER_ONE_SUCCESS,TRIGGER_ONE_FAILED,TRIGGER_NONE_FAILED} trigger_rule_t;// 任务Operatortypedef struct task {char task_id[MAX_TASK_NAME];int (*execute)(struct task *self, void **output);int (*sensor_check)(struct task *self);int is_sensor;int sensor_poke_interval;char *args[8];int arg_count;struct task **upstream_tasks;int upstream_count;struct task **downstream_tasks;int downstream_count;trigger_rule_t trigger_rule;int retries;int max_retries;int retry_delay_sec;int timeout_seconds;task_state_t state;time_t start_time;time_t end_time;char output[1024];struct task *next;} task_t;// DAGtypedef struct dag {char dag_id[64];task_t *tasks;int task_count;char schedule_interval[64];time_t start_date;time_t end_date;int catchup;int max_active_runs;struct dag *next;} dag_t;// DAG Runtypedef struct dag_run {char dag_id[64];char run_id[64];time_t execution_date;char state[16];time_t start_time;time_t end_time;struct dag_run *next;} dag_run_t;// Airflowtypedef struct airflow {dag_t *dags;dag_run_t *dag_runs;xcom_entry_t *xcoms;int dag_count;int xcom_count;pthread_mutex_t mutex;int running;int max_threads;pthread_t scheduler_thread;} airflow_t;2. XCom实现c// 创建Airflowairflow_t *airflow_create(int max_threads) {airflow_t *af malloc(sizeof(airflow_t));memset(af, 0, sizeof(airflow_t));af-max_threads max_threads;af-running 1;af-xcoms NULL;af-xcom_count 0;pthread_mutex_init(af-mutex, NULL);printf([Airflow] 启动最大线程: %d\n, max_threads);return af;}// XCom推送数据void xcom_push(airflow_t *af, const char *task_id, const char *dag_id,const char *key, const char *value) {pthread_mutex_lock(af-mutex);xcom_entry_t *entry malloc(sizeof(xcom_entry_t));strcpy(entry-key, key);strcpy(entry-value, value);strcpy(entry-task_id, task_id);strcpy(entry-dag_id, dag_id);entry-timestamp time(NULL);entry-next af-xcoms;af-xcoms entry;af-xcom_count;pthread_mutex_unlock(af-mutex);}// XCom拉取数据char *xcom_pull(airflow_t *af, const char *task_id, const char *dag_id,const char *key) {pthread_mutex_lock(af-mutex);xcom_entry_t *entry af-xcoms;while (entry) {if (strcmp(entry-key, key) 0 strcmp(entry-dag_id, dag_id) 0) {char *value strdup(entry-value);pthread_mutex_unlock(af-mutex);return value;}entry entry-next;}pthread_mutex_unlock(af-mutex);return NULL;}// 清理旧XCom保留最近100条void xcom_cleanup(airflow_t *af) {pthread_mutex_lock(af-mutex);int count 0;xcom_entry_t *entry af-xcoms;while (entry) {count;entry entry-next;}if (count 100) {xcom_entry_t *prev NULL;entry af-xcoms;int to_delete count - 100;while (to_delete 0 entry) {xcom_entry_t *next entry-next;free(entry);entry next;to_delete--;}af-xcoms entry;}pthread_mutex_unlock(af-mutex);}3. 传感器c// 传感器基类Sensortypedef struct sensor_task {task_t base;int (*check)(struct sensor_task *self);int poke_interval;int timeout;int soft_fail;} sensor_task_t;// 创建传感器任务void dag_create_sensor(dag_t *dag, const char *task_id,int (*check)(sensor_task_t *),int poke_interval, int timeout) {sensor_task_t *sensor malloc(sizeof(sensor_task_t));strcpy(sensor-base.task_id, task_id);sensor-base.is_sensor 1;sensor-base.execute NULL;sensor-check check;sensor-poke_interval poke_interval;sensor-timeout timeout;sensor-base.state TASK_NONE;sensor-base.next dag-tasks;dag-tasks (task_t*)sensor;dag-task_count;}// 传感器示例文件存在检查int file_exists_sensor(sensor_task_t *sensor) {char *filename (char*)sensor-base.args[0];printf([Sensor] 检查文件: %s\n, filename);// 模拟检查实际用access()if (rand() % 3 0) {printf([Sensor] 文件已存在: %s\n, filename);return 1; // 成功}return 0; // 继续等待}// 传感器示例API就绪检查int api_ready_sensor(sensor_task_t *sensor) {printf([Sensor] 检查API就绪...\n);return (rand() % 2 0);}4. 触发规则c// 检查任务是否可执行根据触发规则int task_is_ready_with_trigger(task_t *task) {if (task-upstream_count 0) return 1;int success_count 0;int failed_count 0;int running_count 0;for (int i 0; i task-upstream_count; i) {task_state_t state task-upstream_tasks[i]-state;if (state TASK_SUCCESS) success_count;else if (state TASK_FAILED || state TASK_UPSTREAM_FAILED) failed_count;else if (state TASK_RUNNING || state TASK_SCHEDULED) running_count;}// 如果还有上游在运行不能执行if (running_count 0) return 0;switch (task-trigger_rule) {case TRIGGER_ALL_SUCCESS:return success_count task-upstream_count;case TRIGGER_ALL_FAILED:return failed_count task-upstream_count;case TRIGGER_ALL_DONE:return success_count failed_count task-upstream_count;case TRIGGER_ONE_SUCCESS:return success_count 0;case TRIGGER_ONE_FAILED:return failed_count 0;case TRIGGER_NONE_FAILED:return failed_count 0;default:return 1;}}5. 回填Backfillc// 回填历史数据void airflow_backfill(airflow_t *af, const char *dag_id,time_t start_date, time_t end_date,int max_runs) {printf([Backfill] 回填 DAG: %s, 从 %s 到 %s\n,dag_id, ctime(start_date), ctime(end_date));time_t current start_date;int run_count 0;while (current end_date run_count max_runs) {// 为每个执行时间创建DAG Runchar run_id[64];struct tm *tm_info localtime(current);strftime(run_id, sizeof(run_id), backfill_%Y%m%d_%H%M%S, tm_info);// 创建DAG Run并执行dag_run_t *run malloc(sizeof(dag_run_t));strcpy(run-dag_id, dag_id);strcpy(run-run_id, run_id);run-execution_date current;strcpy(run-state, running);run-start_time time(NULL);run-end_time 0;run-next af-dag_runs;af-dag_runs run;printf([Backfill] 创建执行: %s\n, run_id);// 执行DAG简化提交所有任务dag_t *dag af-dags;while (dag) {if (strcmp(dag-dag_id, dag_id) 0) {task_t *task dag-tasks;while (task) {if (task-state TASK_NONE) {task-state TASK_SCHEDULED;}task task-next;}break;}dag dag-next;}// 前进到下一个执行时间按调度间隔current 86400; // 每天run_count;}}6. 调度器增强c// 调度器主循环增强版void *scheduler_loop_enhanced(void *arg) {airflow_t *af (airflow_t*)arg;while (af-running) {pthread_mutex_lock(af-mutex);dag_t *dag af-dags;while (dag) {// 检查是否有新的执行需要触发回填/调度task_t *task dag-tasks;while (task) {// 处理传感器if (task-is_sensor) {sensor_task_t *sensor (sensor_task_t*)task;static int poke_count 0;poke_count;if (sensor-check sensor-check(sensor)) {task-state TASK_SUCCESS;} else if (poke_count % 5 0) {printf([Sensor] 继续等待: %s\n, task-task_id);}} else if (task-state TASK_NONE || task-state TASK_SCHEDULED) {if (task_is_ready_with_trigger(task)) {task-state TASK_SCHEDULED;// 异步执行pthread_t tid;pthread_create(tid, NULL, (void*(*)(void*))execute_task,(void*)af);pthread_detach(tid);}}task task-next;}dag dag-next;}pthread_mutex_unlock(af-mutex);// 清理旧XComxcom_cleanup(af);sleep(1);}return NULL;}7. 测试代码c// 任务示例生成数据int generate_data_task(task_t *task, void **output) {int value rand() % 100;snprintf(task-output, sizeof(task-output), %d, value);printf([Task] %s 生成数据: %d\n, task-task_id, value);return 0;}// 任务示例处理数据使用XComint process_data_task_xcom(task_t *task, void **output) {// 从XCom获取上游数据// 这里简化使用task输出int data atoi(task-output);int result data * 2;snprintf(task-output, sizeof(task-output), %d, result);printf([Task] %s 处理完成: %d → %d\n, task-task_id, data, result);return 0;}void test_airflow_advanced() {printf( Airflow进阶测试 \n\n);airflow_t *af airflow_create(4);// 创建DAGdag_t *dag malloc(sizeof(dag_t));strcpy(dag-dag_id, data_pipeline);dag-tasks NULL;dag-task_count 0;strcpy(dag-schedule_interval, 0 0 * * *);dag-start_date time(NULL) - 86400 * 3; // 3天前dag-end_date time(NULL);dag-catchup 1;dag-max_active_runs 1;dag-next af-dags;af-dags dag;// 创建任务task_t *t1 malloc(sizeof(task_t));strcpy(t1-task_id, generate_data);t1-execute generate_data_task;t1-is_sensor 0;t1-upstream_count 0;t1-downstream_count 0;t1-max_retries 3;t1-state TASK_NONE;t1-next dag-tasks;dag-tasks t1;dag-task_count;task_t *t2 malloc(sizeof(task_t));strcpy(t2-task_id, process_data);t2-execute process_data_task_xcom;t2-is_sensor 0;t2-max_retries 3;t2-trigger_rule TRIGGER_ALL_SUCCESS;t2-state TASK_NONE;t2-next dag-tasks;dag-tasks t2;dag-task_count;// 设置依赖t1 → t2t1-downstream_tasks malloc(sizeof(task_t*));t1-downstream_tasks[0] t2;t1-downstream_count 1;t2-upstream_tasks malloc(sizeof(task_t*));t2-upstream_tasks[0] t1;t2-upstream_count 1;// 添加传感器dag_create_sensor(dag, wait_for_file, file_exists_sensor, 5, 60);// 设置依赖sensor → t1task_t *sensor dag-tasks;while (sensor !sensor-is_sensor) {sensor sensor-next;}if (sensor) {sensor-downstream_tasks realloc(sensor-downstream_tasks,sizeof(task_t*) * (sensor-downstream_count 1));sensor-downstream_tasks[sensor-downstream_count] t1;t1-upstream_tasks realloc(t1-upstream_tasks,sizeof(task_t*) * (t1-upstream_count 1));t1-upstream_tasks[t1-upstream_count] sensor;}// 回填历史数据time_t start time(NULL) - 86400 * 2;time_t end time(NULL);airflow_backfill(af, data_pipeline, start, end, 3);// 启动调度器af-scheduler_thread pthread_self();scheduler_loop_enhanced(af);// 打印状态printf(\n 任务状态 \n);task_t *task dag-tasks;while (task) {const char *state_str[] {NONE, SCHEDULED, RUNNING, SUCCESS,FAILED, UPSTREAM_FAILED, SKIPPED, UP_FOR_RETRY};printf( %s: %s, task-task_id, state_str[task-state]);if (task-output[0]) {printf( (输出: %s), task-output);}printf(\n);task task-next;}printf(\nXCom条目数: %d\n, af-xcom_count);free(af);}int main() {srand(time(NULL));test_airflow_advanced();return 0;}---三、编译和运行bashgcc -o airflow_advanced airflow_advanced.c -lpthread./airflow_advanced---四、进阶特性对比特性 基础实现 进阶实现任务依赖 ✅ ✅XCom ❌ ✅传感器 ❌ ✅触发规则 ❌ ✅回填 ❌ ✅重试 ✅ ✅超时 ❌ ✅---五、总结通过这篇文章你学会了· XCom任务间数据传递· 传感器等待外部条件· 触发规则灵活控制依赖· 回填历史数据补录· 资源池与连接管理Apache Airflow是数据编排的行业标准。掌握它你就理解了复杂数据管道的调度设计。下一篇预告《从零实现一个分布式数据管道dbt的核心设计》---评论区分享一下你用Airflow编排过什么复杂的数据管道