Convex 调度实战：用 cronJobs 定时任务与 scheduler.runAfter 构建自毁消息应用-育师

数据库
后端

【免费下载链接】convex-backend

The open-source reactive database for app developers

项目地址：https://gitcode.com/gh_mirrors/co/convex-backend

导读

本文以 Convex 官方调度示例应用（npm-packages/private-demos/cron）为主线，系统讲解 Convex 的两大调度能力：由cronJobs()驱动的周期性定时任务（interval/hourly/daily/weekly/monthly 及标准 cron 表达式），以及由scheduler.runAfter驱动的一次性延时任务。读完本文，你将掌握在 Convex 中声明式配置后台定时任务、在 mutation 内动态编排延时逻辑的完整套路，并能结合后端crates/application/src/cron_jobs/mod.rs理解其执行器原理。

一、示例应用总览：一个会"自我销毁"的聊天室

该示例是一个基于 React + Vite 的迷你聊天应用，演示了两类调度场景的典型组合：

周期性后台任务：定时清理消息、重置高分、记录时间戳、发送邮件等，全部声明在convex/crons.ts中，由 Convex 后端自动触发，无需任何外部 cron 服务；
一次性延时任务：发送的每条消息会在 5 秒内逐秒更新倒计时文案，倒计时结束自动删除——实现这个"自毁消息"效果完全依赖scheduler.runAfter，见convex/sendExpiringMessage.ts。

该应用基于 Convex 官方 tutorial 聊天室改造而来（参见 npm-packages/demos/tutorial），在原有"发消息-展示消息"流程之上叠加了调度逻辑，是学习 Convex scheduling 的最小可运行样本。

数据模型

示例仅使用两张表（convex/schema.ts）：

export default defineSchema({ messages: defineTable({ author: v.string(), body: v.string(), }), times: defineTable({ time: v.number(), }), });

messages：聊天消息，自毁逻辑作用于此表；
times：由recordTime定时任务每秒写入一条Date.now()时间戳，供getTimes查询验证定时任务确实在按预期频率执行。

二、运行示例应用

README 给出的启动方式只有两步（在npm-packages/private-demos/cron目录下）：

just install-js npm run dev

just install-js是仓库 Justfile 提供的 JavaScript 依赖安装入口，用于安装本仓库 npm 工作区所需的全部依赖。npm run dev对应package.json中的脚本：

"scripts": { "dev": "convex dev --start 'vite --open'" }

convex dev会同时启动两件事：一是本地 Convex 后端（自动监听convex/目录的代码变更并热更新，包括新增/修改 cron 配置）；二是通过--start 'vite --open'拉起 Vite 开发服务器并自动打开浏览器页面。前端入口页面src/App.tsx用useQuery(api.listMessages.default)实时订阅消息列表，用useMutation(api.sendMessage.default)发送新消息，并展示每条消息的_creationTime。

启动后可以观察两类现象来验证调度生效：

聊天页面发送任意消息，5 秒内消息正文会逐秒追加倒计时并最终消失；
打开times表（或后端日志），可以看到recordTime每秒写入一条记录。

三、周期性定时任务：cronJobs 声明式配置全解析

Convex 的定时任务不需要操作系统级 cron，也不依赖任何外部调度器。你只需在convex/目录下创建crons.ts（注意文件名是固定约定），导出cronJobs()的实例，后端便会自动注册并调度这些任务。示例 convex/crons.ts 完整覆盖了所有内置调度频率：

import { cronJobs } from "convex/server"; import { api } from "./_generated/api"; const crons = cronJobs(); crons.interval("15s task", { seconds: 10 }, api.sendEmail.default); crons.interval( "clear presence data", { seconds: 10 }, api.clearPresence.default, { a: 12, }, ); crons.hourly( "Clear at the top of the hour", { minuteUTC: 0, }, api.clearMessage.default, { n: 1 }, ); crons.daily( "Daily high score reset", { hourUTC: 17, // (9:30am Pacific/10:30am Daylight Savings Pacific) minuteUTC: 30, // no timezone support yet }, api.clearHighScore.default, ); crons.weekly( "Weekly re-engagement email", { dayOfWeek: "tuesday", hourUTC: 17, minuteUTC: 30, }, api.sendEmail.default, ); crons.monthly( "Clear a message once a month", { day: 1, hourUTC: 17, minuteUTC: 30, }, api.clearMessage.default, { n: 1 }, ); crons.cron("clear a message", "0 10 * * 2", api.clearMessage.default, { n: 1 }); crons.cron( "fancier cron job!", "10-30/5 10 1-3 * *", api.clearMessage.default, { n: 1, }, ); crons.interval("record time", { seconds: 1 }, api.recordTime.default); export default crons;

3.1 六种注册方式与参数语义

方法	用途	调度参数示例	触发频率
`crons.interval(name, { seconds }, fn, args?)`	固定间隔重复	`{ seconds: 10 }`/`{ seconds: 1 }`	每 10 秒 / 每 1 秒
`crons.hourly(name, { minuteUTC }, fn, args?)`	每小时整点	`{ minuteUTC: 0 }`	每小时第 0 分钟
`crons.daily(name, { hourUTC, minuteUTC }, fn, args?)`	每天指定时刻	`{ hourUTC: 17, minuteUTC: 30 }`	每天 UTC 17:30
`crons.weekly(name, { dayOfWeek, hourUTC, minuteUTC }, fn, args?)`	每周指定时刻	`{ dayOfWeek: "tuesday", hourUTC: 17, minuteUTC: 30 }`	每周二 UTC 17:30
`crons.monthly(name, { day, hourUTC, minuteUTC }, fn, args?)`	每月指定时刻	`{ day: 1, hourUTC: 17, minuteUTC: 30 }`	每月 1 日 UTC 17:30
`crons.cron(name, "分时日月周", fn, args?)`	标准 cron 表达式	`"0 10 * * 2"`/`"10-30/5 10 1-3 * *"`	见表达式

每个方法统一接收四个参数：

任务名称（唯一标识）：字符串，如"15s task"。修改名称会改变任务的持久化身份，与任务的重建行为相关；
调度配置对象或 cron 表达式：见上表；
目标函数：通过api引用，如api.sendEmail.default——即convex/sendEmail.ts中的default导出；
（可选）传给目标函数的参数：如{ n: 1 }、{ a: 12 }，由 Convex 的校验器（validator）在目标函数执行前进行参数校验。

示例中对同一目标函数api.clearMessage.default注册了多个 cron（hourly、monthly 与两条 cron 表达式任务），说明同一函数可以被多个定时任务复用，只需传不同参数。

3.2 各任务在示例中的实际作用

对照convex/目录下的函数实现：

recordTime（每 1 秒）：recordTime.ts每次向times表插入Date.now()，是最直观的"调度是否在跑"的探针；
sendEmail（每 10 秒 / 每周二）：sendEmail.ts模拟发送邮件（示例中为占位逻辑，实际项目可替换为真实邮件服务调用）；
clearPresence（每 10 秒）：clearPresence.ts清理过期 presence 数据，并演示了带参数调用{ a: 12 }；
clearMessage（每小时 / 每月 / cron 表达式）：clearMessage.ts的defaultmutation 接收{ n: v.number() }，通过db.query("messages").take(n)取前 n 条消息并逐一删除，是"定时清理"的标准范式；
clearHighScore（每天 UTC 17:30）：clearHighScore.ts打印清理日志，演示了时区换算注释——hourUTC: 17, minuteUTC: 30对应太平洋时间 9:30/10:30。

3.3 时区与表达式细节

时区：所有hourUTC/minuteUTC均以UTC为准。源码注释明确写着// no timezone support yet——目前没有时区支持，开发者需自行完成本地时间到 UTC 的换算，这是使用daily/weekly/monthly时必须注意的坑；
weekly的dayOfWeek：取值如"tuesday"（小写英文星期名），示例中以"tuesday"演示；
标准 cron 表达式：格式为分钟小时日月星期五个字段。"0 10 * * 2"表示每周二 10:00 UTC；"10-30/5 10 1-3 * *"演示了范围（10-30）、步长（/5）与多值域（1-3）的组合，即每月 1 至 3 日的 10:10–10:30 之间每 5 分钟触发一次。

四、一次性延时任务：scheduler.runAfter 与"自毁消息"

与周期性 cron 不同，scheduler是注入到 mutation 上下文中的调度句柄，用于在运行时动态编排一次性任务。示例 convex/sendExpiringMessage.ts 给出了一个完整自洽的延时链式调度示例：

// @snippet start self-destructing-message function formatMessage(body: string, secondsLeft: number) { return `${body} (This message will self-destruct in ${secondsLeft} seconds)`; } export default mutation( async ( { db, scheduler }, { body, author }: { body: string; author: string }, ) => { const id = await db.insert("messages", { body: formatMessage(body, 5), author, }); await scheduler.runAfter(1000, api.sendExpiringMessage.update, { messageId: id, body, secondsLeft: 4, }); }, ); export const update = mutation({ args: { messageId: v.id("messages"), body: v.string(), secondsLeft: v.number(), }, handler: async ({ db, scheduler }, { messageId, body, secondsLeft }) => { if (secondsLeft > 0) { await db.patch(messageId, { body: formatMessage(body, secondsLeft) }); await scheduler.runAfter(1000, api.sendExpiringMessage.update, { messageId, body, secondsLeft: secondsLeft - 1, }); } else { await db.delete(messageId); } }, }); // @snippet end self-destructing-message

其执行链路非常清晰，是典型的"递归调度"模式：

用户发送消息，defaultmutation 写入消息并附带(This message will self-destruct in 5 seconds)文案；
立即调用scheduler.runAfter(1000, api.sendExpiringMessage.update, { messageId, body, secondsLeft: 4 })，安排 1 秒后执行update；
1 秒后update被触发：若secondsLeft > 0，用新的倒计时文案db.patch更新消息，并再次runAfter调度自己（secondsLeft - 1），形成每秒一次的链式倒计时；
当secondsLeft归零，走else分支db.delete(messageId)，消息自我销毁。

4.1 scheduler.runAfter 的 API 语义

签名：scheduler.runAfter(delayMs, functionReference, args?)，delayMs单位为毫秒，示例中1000即 1 秒；
只能调度mutation（不能调度 query），因为被调度的函数会执行写操作；
由_generated/api的类型系统保证函数引用与参数类型的编译期校验；
scheduler同样可配合runAt(timestamp, ...)使用（在指定时刻触发），runAfter是其相对时间版本。

4.2 基于 runAfter 的延时清理扩展

clearMessage.ts还导出了一个scheduleClearDatamutation，演示用runAfter实现"N 秒后清理 N 条消息"的延迟批量删除：

export const scheduleClearData = mutation({ args: { n: v.optional(v.number()), }, handler: async ({ scheduler }, { n = 10 }) => { await scheduler.runAfter(n * 1000, api.clearMessage.default, { n: 1 }); }, });

注意这里把延时值换算成毫秒（n * 1000），并在触发时复用同一文件中的default（即批量删除前 n 条消息的 mutation）。这种"调度入口 + 清理执行"的分离设计，与 cron 部分共用目标函数，体现了调度逻辑与业务逻辑解耦的实践。

五、后端执行器原理：CronJobExecutor 如何驱动这些任务

定时任务在 Convex 中由后端进程内的执行器负责"到期即跑"。核心实现位于 crates/application/src/cron_jobs/mod.rs 的CronJobExecutor<RT>（第 111 行起）：

pub struct CronJobExecutor<RT: Runtime> { context: CronJobContext<RT>, running_job_ids: HashSet<ResolvedDocumentId>, /// Some if there's at least one pending job. May be in the past! next_job_ready_time: Option<Timestamp>, job_finished_tx: mpsc::Sender<ResolvedDocumentId>, job_finished_rx: mpsc::Receiver<ResolvedDocumentId>, }

从源码结构可以读出几个关键设计点：

持久化的任务状态：任务元数据（CronJob 定义、下一次运行时间、执行状态/结果）都落在数据库中，模型层由 crates/model/src/cron_jobs 的CronModel管理，涉及CronJob、CronJobState、CronJobStatus、CronNextRun、CronJobResult、CronJobLogLines等类型；
下一次触发时间的计算：通过model::cron_jobs::next_ts::compute_next_ts计算，配合stream_cron_jobs_to_run拉取所有"已到触发时间"的任务；
执行器维护next_job_ready_time：记录"最近一个待执行任务的时间（可能已经过期）"，配合job_finished_tx/rx通道在任务完成后立即重新评估下一轮调度，实现准点触发而非轮询；
任务去重：running_job_ids: HashSet<ResolvedDocumentId>保证同一任务不会并发重复执行；
失败重试与背压：INITIAL_BACKOFF = 500ms、MAX_BACKOFF = 15s提供指数退避，并行度由 knobs 中的SCHEDULED_JOB_EXECUTION_PARALLELISM控制，遇到 OCC 冲突时按UDF_EXECUTOR_OCC_MAX_RETRIES重试；
日志截断：CRON_LOG_MAX_RESULT_LENGTH与CRON_LOG_MAX_LOG_LINE_LENGTH均为 1000，任务结果与日志行会被截断（源码注释说明这些日志仅供 dashboard 展示使用）；
与 ScheduledJobExecutor 的相似性：源码注释指出"这段代码与 ScheduledJobExecutor 非常相似，将来可能重构合并"——也就是说，cronJobs与scheduler.runAfter/runAt最终都汇入同一套持久化调度 + 到期执行的基础设施。

被触发的任务会经由ApplicationFunctionRunner以 mutation 的形式执行（crates/application/src/application_function_runner），并记录函数执行日志，因此你在 dashboard 的"函数日志"中可以看到每次 cron 触发的记录。

六、小结与上手建议

通过这个示例，Convex 调度体系的核心用法可以归纳为三条：

固定周期任务用crons.ts：在convex/crons.ts中export default cronJobs()，按需组合interval/hourly/daily/weekly/monthly/cron六种方式，后端自动持久化调度，无需自建 cron 服务；
运行时一次性任务用scheduler.runAfter/runAt：在 mutation 上下文中动态安排延时执行，配合"链式自我调度"即可实现倒计时、超时清理、延迟通知等场景；
调度与业务解耦：多个 cron 可复用同一函数（如示例反复调用api.clearMessage.default并传不同参数），调度入口只负责"何时触发"，业务函数只负责"做什么"。