@faapi/task-bullmq
faapi 任务子系统的 BullMQ 驱动（Redis 持久化队列）。
基于 Redis 的任务队列——任务不丢（持久化）、多实例天然防重跑（worker 领取制）、支持延迟任务与指数退避重试。适合已有 Redis 基础设施或需要高吞吐的项目。
安装
pnpm add @faapi/task-bullmq bullmq ioredis
bullmq 6 起
ioredis由内置依赖移至 peer dependency——需业务方显式安装（bullmq 5 时代无需）。
## 使用
任务定义不变（主包文件约定 `src/tasks/<name>/task.ts`），只切驱动：
```ts
// faapi.config.ts
import type { FaapiConfig } from '@faapi/faapi';
export default {
task: {
driver: 'bullmq',
bullmq: { connection: { host: '127.0.0.1', port: 6379 } },
// 可选：同 Redis 下多应用隔离
// bullmq: { connection: {...}, prefix: 'myapp' },
},
} satisfies FaapiConfig;
// handler 里用法完全不变
export function POST(body, tasks) {
return tasks.enqueue('send-email', { to: body.email });
}
语义映射
| faapi 驱动接口 | BullMQ |
|---|---|
enqueue(name, payload, { retries, delayMs, dedupId }) | queue.add(name, payload, { attempts: retries + 1, backoff: exponential 500ms, delay, removeOnComplete, removeOnFail, jobId })。终态任务默认 7 天后移除（removeOnComplete/removeOnFail 可覆盖，false 恢复永不清理）——BullMQ 默认永久保留终态任务，Redis 无界增长且 dedupId 去重在存活期内一直生效 |
startWorker(name, { concurrency, process }) | new Worker(name, handler, { connection, concurrency, prefix })；attempt 取 job.attemptsStarted（跨实例/重启准确） |
stop(timeoutMs) | workers.close() + queues.close()（race 超时），超时 abort 在跑任务的 signal |
stopWorkers() | 仅关 Worker（dev 热替换重注册用），Queue 连接保持 |
groups（批量记账，语义见主包 taskBatch.md） | 两组 Redis key：<prefix>:group:<id>（批量行 hash：计数器 + 声明 + mode/nextIndex/payloads/carry）+ <prefix>:group-members:<id>（成员行 hash，jobId→'pending'/'p:'/'s:' 幂等守卫），常驻不自动清理；经 defineCommand/runCommand（BullMQ 类型化 Lua 扩展口，专用连接复用连接池不加新连接）原子计账（settle 同脚本推进 serial 激活指针并解码下一条 payload）。成员批量标识与序号以载荷包装随 data 存储（{ __faapiGroup, __faapiGroupIndex, __faapiPayload }，交付/查询还原）；cancelRemaining 仅对 waiting/delayed 成员 job.remove()，移除成功才落定；settleRemaining 按计数差值补齐落定 |
| 失败重试 | BullMQ 侧执行（attempts = retries + 1，指数退避 500ms 起） |
注意：BullMQ 不提供执行中任务的取消信号，但 stop 超时路径会 abort taskCtx.signal（任务监听 signal 可尽快退出）。
License
MIT