@ eddyq/ 队列

节点.js客户端 埃迪克语name —— 运行的 Rust 任务队列 邮资, (中文(简体) ), 重写或两者兼有.

可以导入三个类:

• Eddyq ——后卫. 交易顺序 耐用批次 迁移.
• EddyqRedis — Redis后端(通过Redis函数). ~70克工作/零散吞食, { every: ms } 时间表,不迁移.
• EddyqApp ——多后端容器. 在一个过程中(如:webhooks → Redis,支付 → Postgres)每行走一条路线。 见 examples/redis-basic/multi.mjs。 。 。 .

见 主读取器 用于后端取舍和基准数,或 @eddyq/wakeboard 为仪表板UI.

安装

pnpm add @eddyq/queue

我们建议 小时 或者说 线条 超过npm——见 npm 备注 为什么NAPI -RS + npm的边缘粗糙.

快开始

import { Eddyq, CancelError, RetryError } from "@eddyq/queue";

const q = await Eddyq.connect(process.env.DATABASE_URL);

q.work("send.email", async ({ payload, id, attempt, signal }) => {
  const res = await fetch(payload.url, { signal }); // signal flips on shutdown

  if (res.status === 429) {
    const after = Number(res.headers.get("retry-after") ?? 60) * 1000;
    throw new RetryError("rate limited", { delayMs: after });
  }
  if (!res.ok && attempt >= 3) {
    throw new CancelError("permanent 5xx; giving up");
  }

  return { bytes: (await res.text()).length };   // stored in eddyq_jobs.result
});

await q.start();

// ... your app runs ...

await q.shutdown(10_000);   // 10s grace; fires abort signals

移徙

eddyq不会自动在靴子上移动 复制的启动。 运行他们作为明确的部署步骤 在启动工人.

货包装有CLI:

DATABASE_URL=postgres://... npx eddyq migrate run
DATABASE_URL=postgres://... npx eddyq migrate list
DATABASE_URL=postgres://... npx eddyq migrate down --max-steps 1 --confirm

或从代码调用:

import { Eddyq } from '@eddyq/queue';
const q = await Eddyq.connect(process.env.DATABASE_URL);
const report = await q.migrate();
console.log("applied:", report.applied.map(m => `${m.version}:${m.name}`));
await q.close();

eddyq.start() 拒绝启动等待迁移。 绕过何时 schema 管理出局,通过 { skipMigrationCheck: true }。 。 。 .

migrate() 拥有能力并拥有 pg_advisory_lock 每个迁移线, 因此运行它两次(或一次从两个部署的主机),安全地序列化.

重试公约

每一次失败都持续到底 { name, message, stack, directive, retryDelayMs } 内 eddyq_jobs.errors[]。 。 。 .

取消

每个处理器都收到 call.signal - 一个 AbortSignal 当火 eddyq.shutdown() 名为. 传去:

q.work("download", async ({ payload, signal }) => {
  const res = await fetch(payload.url, { signal });
  return res.json();
});

shutdown(gracefulTimeoutMs?) 默认为 三万毫秒。 它广播中止到 所有机上操作人员,等待超时,然后强行取消 运行时间任务。 孤儿的工作由心跳扫荡器恢复 下一个幸存的工人.

管理表面

调试

默认是典型工作量的大小——大多数用户应该放过. 把这些递给 start() 当您需要偏离时:

await q.start({
  sweepIntervalMs:        30_000,    // heartbeat sweeper cadence
  staleAfterMs:           60_000,    // when a running job is considered orphaned
  heartbeatIntervalMs:    15_000,    // worker → DB heartbeat (must be ≪ staleAfterMs)
  cleanupIntervalMs:     300_000,    // retention sweep cadence
  completedRetentionSecs: 24 * 3600, // delete completed older than this
  failedRetentionSecs:    7 * 86400,
  cancelledRetentionSecs: 7 * 86400, // pass -1 to keep forever
  leaderLeaseSecs:        30,        // single-leader maintenance lease
  fetchPollIntervalMs:    1_000,     // ignored unless poll-only
});

何时触摸他们:

• 高吞吐量(百万个已完成工作/日) - 降级 completedRetentionSecs {\fn黑体\fs20\shad2\2ah82\3ah20\4ah33\fscx95\3ch592001\be1}所以清扫工作会继续进行.
• 长期操作者(合法操作 > 60s的工作) 提高 staleAfterMs 这样扫荡者就不会找活活儿了. 还要加薪 heartbeatIntervalMs 按比例.
• 许多复制品(50多个) 提高 sweepIntervalMs 所以你没有 敲响每个吊舱的扫描查询.

许可证

麻省理工学院