@ 12- apps/ jobs
类型背景任务——回溯、指数回溯和折叠——后面 可互换驱动器. BULMQ/RDIS 生产、测试中的内置执行以及 零配置开发.
无框架和无域:没有ORM导入,没有主机应用程序类型,没有任务名
属于自己的。 伐木机是一个港口, 租赁的数据库是一个结构接合,
观察是一个端口,也是眼中唯一的网络框架(hono是一个
其子路径背后的可选对等点.
// where the domain lives — the import IS the registration
export const renderReport = defineJob<{ reportId: string }>({
name: "reports.render",
attempts: 5,
backoff: { type: "exponential", delayMs: 5_000 },
handle: async ({ reportId }) => renderAndStore(reportId),
});
// a job that only ever runs on a schedule needs no binding at all
defineJob({
name: "reports.purge",
schedule: { pattern: "*/5 * * * *", timezone: "UTC" },
handle: async () => purgeExpired(),
});
// at an emit site
await renderReport.enqueue({ reportId }, { dedupeKey: reportId });
// at process start — ONE call wires driver, workers, drain, lease and health
import { createApiJobs } from "@12-apps/jobs/server";
import { jobsRouter } from "@12-apps/jobs/hono";
const jobsApi = createApiJobs({
jobs: () => import("./lib/jobs"), // the defineJob modules — required
db: () => getPrismaClient(), // the sweep_leases table (optional)
events: { onJobFailed: pageOnDeadLetter }, // yours, if you want one
});
await jobsApi.start();
app.route("/api/internal/jobs", jobsRouter(jobsApi));
没有 REDIS_URL 而不以任何物配主, start() 解决内线驱动程序
外部生产:操作人员在流程中运行,时刻表不起火(说
因此,应用开始绿色 没有Redis容器。 生产无
Redis 解析到无驱动程序, 加上一个响亮的错误和一个 503 从 /health —— 说
从不坠机,从不沉默的假货. JOBS_WORKER=1 是一个进程
从生产者(仅排队)到消费者(工人+时间表);工人
是同一个图像,不是第二个建筑.
完全收养合同——配置接合,嵌入变量,事件端口
扫地租, 普里斯玛部分, 为什么没有 createWebJobs 已经进入
采用.md。 。 。 .
两项规则
有效载荷携带识别器,永远不能状态. { reportId },而不是提供
报告。 数据库是真理的来源; 队列只决定 何时A级
重复一行的有效载荷是第二份副本,可以不同意,而且
是复制品在几天后被采取行动.
掌者为一相. 分娩时间最少——工人可能死亡 在副作用和承认之间。 数据库里独有的优势 限制,而不是排队.
它拒绝什么
3个线接错误被拒绝而不是被吸收,因为每个错误都 否则 无声 - 过程开始,探测器是绿色的,并且工作 简单停止发生:
- 一个无法运行的工作.
defineJob丢入空白名称, 空queue(它永远不会回到默认状态——它创造了一个队列无人 非阳性attempts编号 :concurrency毫无延迟的后退, 和一个不是5到6个字段的克龙图案."@daily"财务报告和审定财务报表""两者 安装永不起火的调度器. - 没有工作的部署.
createApiJobs({ jobs: [] })掷出 工厂; ajobs收音机里没有投出的东西start(); 和 :startJobWorkers()丢入空的注册簿, 因此手接根路径 被同样看守。 宣布什么也不是关闭工作的方法——JOBS_DRIVER=off没错. - 一个没有工人可以要求的排队.
enqueue返回时{ enqueued: false, reason: "unregistered" }登记处 不以这个名字为名(身份,不只是名字)。 备选案文 是后端接受的书写,没有处理者索偿,消费者 死信——在打电话者被告知之后enqueued: true。 。 。 . - A 类
retention窗口. 每个数字都必须是正数 和有限的。 负数count不保留更少的工作,它每修剪一个 完成而不是维持上限;NaN(未设置的 env var 读取通过)Number()) 完全禁用比较。 在工厂被拒绝 又在驾驶室里.
担保和非担保
enqueue从不扔。队列退出返回{ enqueued: false }财务报告和审定财务报表 日志; 它不辜负推迟工作的请求.- 后端内存:完成的工作保留一天,失败的一周,
和具有不同支持窗口的主机通过自己的
retention- 谁是 由于负数或负数NaN窗口停止边框 后端 而不是缩小它. - 附表为 已核对 从代码中删除的 cron 任务有其
从Redis取出调度器,而不是永远向一个没有的处理器开火
更长久地存在。 报告已删除
events.onScheduleRemoved, (中文(简体) ), 因为它是破坏性的,是部署完成的. - 一个停滞的工作被重新运行,并报告. 当工人锁上工作
过期( 被屏蔽的事件循环、 被暂停的进程、 无声的 Redis 链接), BullMQ
把工作放回去再运行它。 每个摊位都是错误的
events.onJobStalled。一个持续拖延的工作(超过maxStalledCount时间,或者,对于一个预定的工作, 更多的开始比它的尝试加那)是 作为终端死信失败, 而不是被永久重新启动 。 这个 锁定和摊位数字可按队列排列stall。 。 。 . - 这个
inline驾驶员荣誉attempts但不拖延或时间安排,以及两者 遗漏被记录下来而不是沉默。 生产中拒绝这样做—— 名字, env var 和实例.
BullMQ 驱动程序导出自 @12-apps/jobs/bullmq从来没有枪管,所以
导入 defineJob 在放电现场不会把Redis拖入捆绑.
createApiJobs 保持同样的地产:它懒惰地进口了公牛MQ驱动程序,
只有在一个过程中, 决议实际上选择了它.
Redis 必须运行与 maxmemory-policy noeviction - 司机检查和
投诉.
告诉别人发生了什么事
这包不通知任何人,不审计,不公布任何事件。 或 时刻是港口:
createApiJobs({
jobs: () => import("./lib/jobs"),
events: {
// The dead-letter. `terminal` is the whole point — a non-terminal failure
// is ordinary retry noise.
onJobFailed: ({ name, error, terminal }) => {
if (terminal) void notifyOperators(name, error);
},
onJobCompleted: ({ name, runId }) => void publishRealtime(name, runId),
// A job lost its lock mid-run and was put back to run again.
onJobStalled: ({ name, runId, stalledCount }) => void notifyOperators(name, runId, stalledCount),
// A deploy just cancelled a recurring job, permanently.
onScheduleRemoved: ({ name, queue }) => void audit("schedule.removed", { name, queue }),
},
});
投出或拒绝的观察者会被记录下来并吞咽:别人的 代码不能失败它正在监视的工作 .
扫荡租约
已排定的扫描声明 queue: SWEEP_QUEUE, concurrency: 1,这就
他们在一个工人内部单飞。 整个复制品需要保证
在DATABASE中指明的、有时限的索赔 -- -- createSweepLease (或约束
withSweepLease 编辑 createApiJobs 返回 :
defineJob({
name: "reports.purge",
queue: SWEEP_QUEUE,
concurrency: 1,
schedule: { pattern: "0 * * * *" },
handle: async () => {
const { ran } = await jobsApi.withSweepLease("reports.purge", 10 * 60_000, () =>
purgeExpired(),
);
if (!ran) return; // another worker holds this tick
},
});
索赔是有条件的更新——数据库选择胜者,并且只有
丢失的种族是沉默的; 丢失的桌子或死店 THROWS, 所以停了
扫荡工作失败了 这个 SweepLease 带这个的桌子船
软件包( r)prisma/jobs.prisma + 键 prisma/migrations/)并同步到
宿主的计划; 部分和迁移是 COPIED , 从未连结 .