@absolutejs/队列
持久且已打入的背景工作队列 爱莉西亚 编辑 绝对JS生态系统. 坚持工作,在工人之间安全地要求工作 回放,死信,运行延迟一发.
没错 没有 重塑克龙 配对
@elysiajs/cron 用于反复触发。 克龙
决定 何时; 队列保证工作 发生 (一旦成功重启).
状况:早(
0.4.0) (中文(简体) ). 内部商店,有图案定义的打字登记册,工人, Elysia 插件、管理路线、独立工人跑道、arunHandlerOnce手动触发/测试的助手,以及控制平面createWakeScheduler被闲置的房客叫醒 生产商店:@absolutejs/queue-postgres。 。 。 .
安装
bun add @absolutejs/queue elysia
使用量
import { Elysia } from 'elysia';
import {
createInMemoryJobStore,
createJobRegistry,
defineJobs,
queue,
t
} from '@absolutejs/queue';
// Define jobs once: kind -> payload schema. Payload types are inferred from this
// (no hand-written job map, no generics) and validated at enqueue + dequeue.
// Build schemas with `t` from this package so they share one TypeBox instance.
const jobs = defineJobs({
'email.send': t.Object({ to: t.String(), subject: t.String() }),
'webhook.deliver': t.Object({ url: t.String(), body: t.Unknown() })
});
const store = createInMemoryJobStore(jobs);
const registry = createJobRegistry(jobs)
.on('email.send', async ({ to, subject }) => {
// to: string, subject: string — inferred from the schema
})
.on('webhook.deliver', async ({ url, body }, { attempts }) => {
// retried automatically; `attempts` is which try this is
});
const app = new Elysia()
.use(queue({ registry, store })) // in-process worker auto-starts
.post('/welcome/:email', ({ params, queue }) =>
queue.enqueue('email.send', {
subject: 'Welcome',
to: params.email
})
)
.post('/notify', ({ body, queue }) =>
// delayed one-shot: deliver the webhook in 1 hour
queue.enqueue(
'webhook.deliver',
{ body, url: 'https://example.com/hook' },
{ runAt: Date.now() + 60 * 60 * 1000 }
)
)
.listen(3000);
经常性工作(与 @elysiajs/cron(中文(简体) )
模式:保留 store 在模块范围, 所以两个队列插件的工人
和克龙触发 引用相同的支持状态。 弯曲 run
召回没有爱丽西亚 Context,所以无法达到队列
它关闭在进口的 store 直接说.
// src/jobs/index.ts
import {
createInMemoryJobStore,
createJobRegistry,
defineJobs,
queue,
t
} from '@absolutejs/queue';
import { cron } from '@elysiajs/cron';
import { Elysia } from 'elysia';
const jobs = defineJobs({
'email.send': t.Object({ to: t.String(), subject: t.String() })
});
// Module-scoped so cron + worker reference the same backing state.
export const store = createInMemoryJobStore(jobs);
export const registry = createJobRegistry(jobs).on(
'email.send',
async () => {}
);
export const backgroundJobs = new Elysia({ name: 'background-jobs' })
.use(queue({ registry, store }))
.use(
cron({
name: 'weekly-digest',
pattern: '0 8 * * 1', // Mondays at 08:00
run: () =>
store.enqueue({
idempotencyKey: `weekly-digest:${new Date().toISOString().slice(0, 10)}`,
kind: 'email.send',
payload: {
subject: 'Weekly digest',
to: '[email protected]'
}
})
})
);
标出每日队列 idempotencyKey 所以误射不会双跑.
醒来被闲置杀害的租户(createWakeScheduler(中文(简体) )
在PaaS, 租户过程被闲置杀害 正在处理中的时间器。 每6小时触发一次的客户圈,如果 房客睡了五个人 修补是建筑: 控制平面 (永不被闲杀) 拥有时间表和插孔 租户醒了;租户自己排队的工人然后做实际工作.
createWakeScheduler 是控制飞机的一面: 一个持久的排程器
按租户计的火every 间隔或5-战地 UTC cron)和呼叫
呼叫器供应 wake 动作——HTTP插孔, runtime.ensure(tenant), (中文(简体) ),
不管是什么把房客带来.
import { createWakeScheduler, httpWake } from '@absolutejs/queue';
// Runs on the CONTROL PLANE (always-on), not in tenant processes.
const wakeScheduler = createWakeScheduler({
entries: [
// Exactly one of `every` (ms) or `cron` (5-field, UTC) per entry.
{
every: 6 * 60 * 60 * 1000,
id: 'acme-digest',
tenant: 'acme',
url: 'https://acme.internal/wake'
},
{ cron: '0 8 * * 1', id: 'globex-report', tenant: 'globex' }
],
// The wake is the single seam. httpWake() POSTs entry.url and treats
// non-2xx as an error...
wake: httpWake(),
onError: (error, entry) => console.error(entry?.id, error)
});
wakeScheduler.start();
与 @absolutejs/runtime, (中文(简体) ),
唤醒是过程复活—— ensure 给房客穿靴子
租户的排队工人在起步时 接应的工作:
const wakeScheduler = createWakeScheduler({
entries: tenantSchedules,
wake: async (entry) => {
await runtime.ensure(entry.tenant);
}
});
调度员保证:
- 重新启动安全 —— 说
snapshot()编号 :restore(snap)持续条目 加每入数lastFiredAt(和记忆店的一样) 快照/恢复),所以控制平面既未重启双发又未重启 失去时间表.catchUp: 'skip'(默认) 空投 休息时间和恢复节奏;catchUp: 'once'单发 补偿性醒悟。 不管怎样metrics().missedSkipped数起他们. - 错误隔离 ——同时唤起起起火来;有一位房客失眠了
点击数
onError+该errors永远不要挡住其他的 失败 唤醒等待它的下一个插槽(唤醒是一个戳子,而不是一个工作——没有后退) (原始内容存档于2018-09-31). entertainment. - 没有双发 每对接分钟最多开一炮; 重复的勾勾被跳过和计算, 永远不要运行两次 .
- 操作员表面 —— 说
add编号 :remove编号 :enable编号 :disable编号 :list, (中文(简体) ),metrics()(单位:千美元)entries, (中文(简体) ),enabled, (中文(简体) ),firings, (中文(简体) ),errors, (中文(简体) ),missedSkipped, (中文(简体) ),skippedTicks, (中文(简体) ),lastTickMs, (中文(简体) ),byTenant),以及drain()与worker.drain()。 。 。 .tick()用注射器进行试验时暴露now· ;jitterMs通过可注射器传播同级机队random。 。 。 . - 追踪 - 带有
tracerProvider每一次射击都是queue.wake带宽abs.tenant+ 键abs.wake.id匹配queue.runJob。 。 。 .
在repo cron 解析器中故意最小化: 5个字段
(单位:千美元)minute hour day-of-month month day-of-week, (中文(简体) ), *数字 逗号列表,
范围, /n 步骤 * 或范围,星期日 0-7 (两者) 0 财务报告和审定财务报表 7
是星期天),标准日-日-场-比对语义学, 仅协调世界时。无
名称 (MON),无 L编号 :W编号 :#没有秒,没有宏——如果你需要这些,
运行一个真正的克龙和打电话 wake 你自己来.
一发手动扳机runHandlerOnce(中文(简体) )
有时你想直接引用一个处理器——手动回填,管理员
重新运行,单位测试,或 bun scripts/foo.ts 共享逻辑的包装符
颈部。 使用 runHandlerOnce: 它通过
登记册的图和合成a JobContext 为你,然后运行
处理者——没有工人,没有商店写作.
// scripts/runWeeklyDigest.ts
import { runHandlerOnce } from '@absolutejs/queue';
import { registry } from '../src/jobs/registry'; // direct import — see warning below
await runHandlerOnce(registry, 'email.send', {
to: '[email protected]',
subject: 'Weekly digest (manual trigger)'
});
不要进口再出口的枪管
backgroundJobs。 。 。 . 正在导入 Elysia 插件调用@elysiajs/cron使计时器活着 防止您的脚本退出。 (a) 将工作模块拆分为:registry从不同的文件中导出backgroundJobs,或 (b) 国家process.exit(0)在剧本结尾处.
runHandlerOnce 接受一个 options.context 覆盖(用于 attempts, (中文(简体) ),
maxAttempts, (中文(简体) ), id(等等)和a options.validators 覆盖( r)false 改为
跳过校验, 或预编译 JobValidators 热回路.
如何运作
- Schema 定义的任务 —— 说
defineJobs是真理的唯一来源:有效载荷 类型从TypeBox schemas中推断(没有手写任务图,没有)<Jobs>(b) 有效载荷在排队和排队时经过验证. - 已打入的登记册 —— 说
kind → payload → handler,检查端到端. JobStore接口 —— 说enqueue, (中文(简体) ),claimDue(原子),complete, (中文(简体) ),fail, (中文(简体) ),reapStuck, (中文(简体) ),listByKind交换createInMemoryJobStore对于一个耐久的适配器在prod.- 工人 - 要求工作达到货币上限、操作操作、重复
指数后退, 死字母
maxAttempts并收获工人的工作 死亡(租赁+reapStuck) (中文(简体) ). - 目 录 - 通过
idempotencyKey改为enqueue来去去dupe.
许可证
BSL-1.1 — 见 自由主义。根据许可证条款在更改日期上转换为 Apache 2. 0.
索赔所有权和过期工人
仓库可以曝光 completeClaim(id, claimToken) 财务报告和审定财务报表
failClaim(id, claimToken, options)返回时 false 当尝试没有
长期拥有声称的工作。 工人在
这两家商店都在场,每件索赔都提供一新代币.
使用 requireClaimFencing: true 用于必须拒绝遗留遗产的持久加工
储存没有原子索赔. JobContext.claimToken 让处理者围栅
应用检查点/结果写作反对同样的尝试.
此保护队列状态 。 它不会消除外部影响 也不会阻止处理器
这忽略了取消。 帮手必须尊重 signal 并使用应用程序
所有权/尝试性检查或持久效果收据.
过期的债权在收割时需要一次尝试。 达到 maxAttempts 移动
任务到 dead 有租期错误,而不是无限期重试.
不同工作管理人员的工人
工人现在将注册的操作人员类别传给 claimDue. 商店广告
supportsKindFiltering: true 必须在解剖时应用过滤器 在此之前 页:1
索赔限额; kinds: [] 声称什么。 内幕商店支持这一点
队列后置0.1.7 对其职务定义的额外限制。 启用
requireKindFiltering: true 拒绝一个无视合同的适配器.
自定义适配器必须在广告支持前执行过滤.
在引入新的工作类型之前, 提高每个工人的排队水平 。 一个老人 工人仍然可以消费它不承认的工作,直到升级或 已停止; 仅对一个生产者进行升级无法改变这个旧的流程.
部署排出量
打电话 worker.drain() 在退休之前, 它停止了新的索赔,而现有的索赔和处理者则结束。 等两个 worker.metrics().active 财务报告和审定财务报表 worker.metrics().claiming 在终止进程前为零. worker.stop() 等待待决的民意调查、数据库索偿和处理者本身。 如果部署失败了, worker.resume() 重新申请许可。 尚待处理的索赔准备货币,因此,重叠的民意调查不能使工人填满.