Node 后端实战 · 边缘 Cron 定时任务怎么写？Cloudflare 三个实战任务与踩坑

Node 后端实战 · 边缘 Cron 定时任务怎么写？Cloudflare 三个实战任务与踩坑

各位看官，把定时任务跑在边缘网络上，和我以前在单机crontab或容器里写个@Scheduled是完全两码事。以前那套心智模型是「一台机器，一个进程，准点跑一次」，但 Cloudflare Workers 的 Cron Triggers 是「每天某个时刻，平台在世界各地的边缘节点挑一个触发你的 Worker」——你不知道它跑在哪、上一秒的实例还在不在、这一次要跑多久会被掐。

我这个多租户系统有四个每天凌晨跑的定时任务：预聚合统计、拦截名单对账、审计日志冷热归档、导出文件清理。这篇文章把它们的真实实现拆开讲，重点不是「怎么调 API」，而是边缘环境逼出来的那几个设计取舍和真实踩过的坑。

一、Cron Triggers 长什么样

配置即代码，写在wrangler.toml里，dev 本地不触发，只有 test/prod 注册：

# ---- Cron Triggers（仅 test/prod 注册，dev 本地不触发）---- [env.prod.triggers] crons = ["0 0 * * *", "0 1 * * *", "0 2 * * *", "0 3 * * *"]

四个 cron 表达式，分别对应四个任务。入口是 Worker 的scheduled钩子，它把事件分发到我的任务处理器：

// src/index.tsasyncscheduled(event:ScheduledEvent,env:Bindings,ctx:ExecutionContext){dispatchCron(env,"stats-aggregate").then((r)=>{/* 0 点 */});dispatchCron(env,"blocklist-reconcile").then((r)=>{/* 1 点 */});dispatchCron(env,"audit-sweep").then((r)=>{/* 2 点 */});dispatchCron(env,"export-cleanup").then((r)=>{/* 3 点 */});}

这里第一个要建立的认知：Cron Triggers 是「尽力而为」的，不是精确 cron。平台正常情况下每天触发一次，但极端情况下可能漏跑，也可能（极少见地）重跑。所以我的每一个任务都设计成幂等——重跑不会重复写脏数据。后面会反复看到这一点。

四个任务的全貌先放一张表，后面对着看更清楚：

二、通用骨架：日志 + 锁 + 错误脱敏

四个任务共用一套骨架，核心在dispatchCron里：

exportconstdispatchCron=async(env:Bindings,task:string)=>{constdb=getDb(env);// 分布式锁：KV 读-写非原子（KV 没有 CAS），纯并发下两个调用可能同时读到空都进入执行。// Cloudflare Cron 正常不会重复触发，此为 best-effort 防护，非严格互斥。constlockKey=`cron:lock:${task}`;constlockVal=awaitenv.KV.get(lockKey);if(lockVal==="running")return{ok:true,processed:-1,error:"already running"};awaitenv.KV.put(lockKey,"running",{expirationTtl:600});const{logId,startedMs}=awaitcronStart(db,task);// 写 cron_exec_logstry{// ...按 task 名分发...awaitcronEnd(db,logId,startedMs,processed,true);return{ok:true,processed};}catch(e:unknown){constraw=einstanceofError?e.message:String(e);constmsg=raw.length>120?raw.slice(0,120)+"...":raw;// 脱敏：截断+移除 SQL/路径awaitcronEnd(db,logId,startedMs,0,false,msg).catch(()=>{});return{ok:false,error:"cron task failed"};}finally{awaitenv.KV.delete(lockKey).catch(()=>{});}};

三个设计点值得单独拎出来：

1. 执行日志cron_exec_logs：每个任务首尾都写一条记录（开始时间、结束时间、处理条数、耗时、状态、错误信息）。定时任务最怕「静默失败」——你以为它跑了，其实半路挂了。有了这张表，运维直接查status='failed'就能知道哪天哪次挂了，而不是靠用户投诉才发现。
2. KV 分布式锁是 best-effort：锁用KV.put(key, "running", { expirationTtl: 600 })，600 秒自动过期兜底。但注释里写得很诚实——KV 没有 CAS（比较并交换），读-写非原子，理论上并发时两个调用都能读到空值、都进入执行。Cloudflare Cron 正常不会重复触发，所以这把锁只是兜底，不能当严格互斥用。真要强一致互斥得用 Durable Objects，但为了防一个几乎不会发生的重跑去引入 DO，不划算。
3. 错误信息脱敏：异常堆栈里可能含 SQL 语句、表名、路径，直接落库有信息泄露风险。所以截断到 120 字符，对外只回cron task failed，细节进日志。这条和我在审计日志里的脱敏思路一致。

三、任务一：stats-aggregate 预聚合统计

这个任务每天凌晨把前一天的实时数据聚合成一张宽表（lead_stats_daily），接口查统计时直接读这张预聚合表，而不是每次对大表GROUP BY。

为什么必须预聚合：实时统计要同时算「按状态分布、按分类分布、按项目分布、当日新增、当日跟进、当日转化、通话统计、坐席绩效」……这些如果每次请求都现算，大表上一堆GROUP BY直接把接口拖垮。每天算一次、结果落表，读接口从 O(聚合扫描) 变成 O(1 主键查)。

核心难点是「一次算全」，我用Promise.all把七个无依赖的聚合查询并行发出去，而不是串起来等：

// 并行执行无依赖的聚合查询（DATA-10）const[byStatusRows,catRows,projRows,addedRow,fuRow,convRow,callStatRow]=awaitPromise.all([db.select({status:leads.status,n:count()}).from(leads).where(and(eq(leads.tenantId,tid),isNull(leads.deletedAt))).groupBy(leads.status),// ...按分类、按项目、当日新增、当日跟进、当日转化...db.select({callCount:count(),answeredCount:sql<number>`sum(case when${callRecords.answerType}= 'answered' then 1 else 0 end)`,noAnswerCount:sql<number>`sum(case when${callRecords.answerType}!= 'answered' then 1 else 0 end)`,}).from(callRecords).where(/* 时间窗 + 租户隔离 */),]);// 坐席绩效：用 GROUP BY 聚合查询替代「循环内每条查一次」的 N+1 模式（DB-07）constuserRows=awaitdb.query.users.findMany({where:/* 租户内 */,columns:{id:true,name:true}});constfuAgg=awaitdb.select({userId:leadFollowups.userId,cnt:count()}).from(leadFollowups).where(/* 时间窗 + 租户 */).groupBy(leadFollowups.userId);// ...通话聚合、转化聚合、最近跟进时间 同理 GROUP BY，再用 Map 在内存里按 userId 拼装...

两个真实优化点：

• Promise.all并行：七个聚合查询之间没有依赖，串行会累积延迟，并行把总耗时压到最慢那一个。
• GROUP BY 替代 N+1：坐席绩效如果按「先查用户列表、再循环为每个用户发一条查询」写，就是经典的 N+1。我改成几条GROUP BY聚合 + 内存Map拼装，一次扫全表而不是 N 次。

幂等 upsert：任务支持手动重跑——如果某天数据算错了，触发一次补算不会插重复行，而是覆盖：

constexisting=awaitdb.query.leadStatsDaily.findFirst({where:and(eq(leadStatsDaily.tenantId,tid),eq(leadStatsDaily.statDate,dateStr)),});if(existing){awaitdb.update(leadStatsDaily).set({/* 全部字段 */}).where(eq(leadStatsDaily.id,existing.id));}else{awaitdb.insert(leadStatsDaily).values({id:crypto.randomUUID(),/* 全部字段 */});}

一个真实的坑（必讲）：聚合查询里sum(case when ... then 1 else 0 end)这种，如果当天的callRecords一条都没匹配上（空集），SQLite 的sum()会返回NULL，而不是 0。而我的表字段是NOT NULL。我第一次跑的时候，直接callStat.callCount当数字用写进去，结果触发NOT NULL约束报错、整批失败。后来改成逐字段?? 0兜底——注意?? 0只兜底「缺行」，不兜底「行内有 NULL 字段」，所以每个answeredCount / noAnswerCount都得单独兜底。这种边缘 case 不跑一次真发现不了。

四、任务二：blocklist-reconcile 拦截名单对账

这个任务每天扫描全量数据，根据最新的拦截名单刷新每条记录的「是否被拦截」标记，并联动取消其待执行的计划。

用 Set 消除 N+1：最蠢的写法是对每一条记录去查一次「它在不在拦截名单里」。正确做法是先把名单一次性查出来建一个Set，然后 O(1) 判断：

constblRows=awaitdb.query.blocklist.findMany({where:and(isNull(blocklist.deletedAt),or(eq(blocklist.scope,"platform"),and(eq(blocklist.scope,"tenant"),eq(blocklist.tenantId,tid)))),});constblockedPhones=newSet(blRows.map((r)=>r.phone));// 平台级 + 租户级名单合并letcursor=0;for(;;){constbatch=awaitdb.query.leads.findMany({where:and(eq(leads.tenantId,tid),isNull(leads.deletedAt),gte(leads.createdAt,cursor)),orderBy:[asc(leads.createdAt)],limit:RECONCILE_BATCH,// 1000});if(batch.length===0)break;for(constleadofbatch){if(!lead.phone)continue;consttarget=blockedPhones.has(lead.phone)?1:0;if(target!==lead.isBlocked){// 仅值变更才写，避免无谓写放大awaitdb.update(leads).set({isBlocked:target}).where(eq(leads.id,lead.id));if(target===1){awaitdb.update(schedules).set({status:"cancelled"}).where(and(/* 该线索的待执行计划 */));}affected++;}processed++;}if(batch.length<RECONCILE_BATCH)break;cursor=batch[batch.length-1]!.createdAt+1;// 游标续跑}

两个细节：

• 游标分页替代 OFFSET：大表用OFFSET深翻页会越来越慢，我用createdAt游标（where createdAt >= cursor），每批 1000 条，批次末尾的createdAt+1作为下一批起点，断点可续、深翻页稳。
• 只写变更：target !== lead.isBlocked才 UPDATE，绝大多数记录标记没变，省下大量写操作。

五、任务三：audit-sweep 审计日志冷热分层归档

审计日志只增不删，时间一长 D1 存储和查询都扛不住。这个任务做分层清理：常规操作保留 365 天，关键操作（导出、擦除、重置密码、登出全部设备等）保留 1825 天（5 年），过期且非关键的转存到 R2 冷归档后从 D1 删除。

原子性铁律——先归档成功，才删源数据：这是整个系统我立得最死的一条规矩。R2 写入失败，宁可跳过这一批、绝不删 D1，绝不能「源没了归档也没成」导致数据丢失：

consthotThreshold=Math.floor(Date.now()/1000)-AUDIT_HOT_DAYS*86400;// 365 天constcriticalThreshold=Math.floor(Date.now()/1000)-AUDIT_CRITICAL_DAYS*86400;// 1825 天constcriticalActions=["lead.export","lead.erase","customer.erase","user.reset-password","logout-all","tenant.suspend","tenant.renew","user.create"];for(;;){constbatch=awaitdb.select().from(auditLogs).where(and(gte(auditLogs.createdAt,cursor),or(and(not(inArray(auditLogs.action,criticalActions)),lt(auditLogs.createdAt,hotThreshold)),and(inArray(auditLogs.action,criticalActions),lt(auditLogs.createdAt,criticalThreshold)),))).orderBy(asc(auditLogs.createdAt)).limit(SWEEP_BATCH);// 1000if(batch.length===0)break;for(constrowofbatch){constndjsonLine=JSON.stringify(row)+"\n";constkey=`audit-archive/${row.tenantId??"_platform"}/${month}/${row.id}.ndjson`;try{awaitenv.BUCKET.put(key,ndjsonLine);// 先写 R2 冷归档awaitdb.delete(auditLogs).where(eq(auditLogs.id,row.id));// 成功后才删 D1totalArchived++;totalDeleted++;}catch(e){console.error(`[audit-sweep] R2 put failed for${row.id}:`,e);totalSkipped++;// R2 失败 → 跳过该条，绝不删源}}if(batch.length<SWEEP_BATCH)break;cursor=batch[batch.length-1]!.createdAt+1;}

为什么这条铁律重要：如果反过来「先删 D1 再写 R2」，一旦 R2 那一下网络抖了或超限，这条审计记录就永久消失了——而审计日志在很多场景下是合规刚需，丢了是要出事的。归档和删除之间如果不保证顺序，就是在赌 R2 永远不出错。所以顺序必须「先 R2 成功，再删 D1」，R2 挂了就留着 D1 等下轮重试。

month用toISOString().slice(0, 7)取YYYY-MM，归档路径按租户+月份分目录，方便后续按时间检索或整体删桶。

六、边缘 Cron 的几个心智模型

写完三个任务，回头总结几条边缘 Cron 和单机 cron 最大的不同：

所以边缘定时任务的设计主线就两条：幂等（重跑不脏数据，靠 upsert/游标续跑）+分批（大表游标扫、每批 1000 条、R2 失败不删源）。把这两条焊死，漏跑重跑都不怕。

七、小结

边缘 Cron 不是「把 cron 表达式搬上云」那么简单。它逼你重新想清楚：状态放哪（KV/R2/D1）、会不会重跑（幂等 upsert）、大表怎么扫（游标分批）、跨存储操作怎么不丢数据（先归档后删）。我这系统的四个任务，就是用上面那些真实代码一点点磨出来的——尤其是审计归档的原子性铁律和聚合查询的 NULL 兜底，都是线上真踩过才长记性的。

如果你的系统也在用 Cloudflare Workers，定时任务这块建议从第一天就把「执行日志表 + 幂等 + 分批」当成标配，别等半夜被报警叫起来才发现任务静默失败了。

相关阅读

• Node 后端实战 · 后端敏感数据怎么防泄露？PII 自动脱敏与审计日志实战
• Node 后端实战 · Cloudflare Workers 限流总误伤？用内存固定窗口替代 KV 实战
• Serverless 导出 CSV 总超时？用 Queue + R2 异步任务彻底解决
• Node 后端实战 · 多租户 SaaS 的数据隔离
• Node 后端实战 · JWT 双密钥轮转与 token 版本号
• Node 后端实战 · D1 那些坑
• Node 后端实战 · 架构决策全景
• Koa 实现 JWT 会话与鉴权，前后端分离项目通用方案
• MySQL 生产环境备份与恢复完整方案

本文由 FungLeo 主导，Deepseek 优化校阅，转发请注明首发地址，谢谢大家！