Bullmq

BullMQ 队列任务开发指南

技术栈

• 队列系统: BullMQ v5.16.0+
• 类型安全: TypeScript + Zod 运行时验证
• Effect 集成: Effect 系统封装（Promise → Effect）
• Redis: ioredis (maxRetriesPerRequest: null)
• UI 监控: Bull Board (Hono adapter)

文件结构

src/lib/
├── enums/bullmq.ts                      # 队列和任务名称枚举
├── infrastructure/
│   ├── bullmq/
│   │   └── job-registry.ts              # 类型映射和 Zod 验证
│   ├── bullmq-adapter.ts                # QueueManager 核心类
│   └── effect/services/bullmq.ts        # Effect Layer
│   └── bootstrap.ts                     # Worker 注册位置
src/routes/admin/
└── queue-board.index.ts                 # Bull Board UI 路由

核心规则（MANDATORY）

三层类型安全架构

1. 编译时约束：QueueJobsMapping 确保队列只能使用特定 job name
2. 类型推断：JobDefinitionRegistry 自动推断 job data 类型
3. 运行时验证：JobSchemaRegistry 使用 Zod 验证数据

命名约定

• 队列名称：小写字母（email、cleanup，不用 EMAIL_QUEUE）
• 任务名称：kebab-case（send-welcome、daily-cleanup）
• 常量枚举：对象字面量 + as const（不用 enum）
• 索引签名：[JobName.XXX]: Schema 形式

数据验证规则

• 所有 job data 必须有对应的 Zod schema
• Schema 必须包含中文错误消息
• 日期格式：ISO 8601 字符串（2024-01-01T00:00:00Z）
• 日期字符串：YYYY-MM-DD 格式
• UUID：使用 z.uuid() 验证

Effect 使用规范

• 所有异步操作使用 Effect 封装（Effect.tryPromise）
• 同步操作使用 Effect.sync
• 错误统一返回 Error 类型
• Worker 注册使用 Effect.sync（立即返回 Worker 实例）

开发步骤

1. 添加新队列

参考 queue-definition.md

1. 在 src/lib/enums/bullmq.ts 添加队列名称
2. 在 job-registry.ts 创建 QueueJobsMapping 映射（初始为 never）
3. 注册 Worker（见步骤 3）

2. 添加新任务类型

参考 job-definition.md

1. 在 src/lib/enums/bullmq.ts 添加任务名称
2. 在 job-registry.ts 创建 Zod schema
3. 更新 JobDefinitionRegistry 类型映射
4. 更新 JobSchemaRegistry 验证映射
5. 更新 QueueJobsMapping 关联队列

3. 注册 Worker

参考 worker-registration.md

1. 在应用启动时调用 queueManager.registerWorker
2. 实现 processor 函数（接收 Job<T> 类型）
3. 可选配置：并发数、限流、重试策略
4. Worker 自动验证 job data（无需手动验证）

4. 添加任务到队列

在业务代码中使用（无需创建文件）：

import { Effect } from "effect";
import { queueManager } from "@/lib/infrastructure/bullmq-adapter";
import { JobName, QueueName } from "@/lib/enums/bullmq";

// 添加任务
const program = queueManager.addJob(
  QueueName.EMAIL,
  JobName.EMAIL_SEND_WELCOME,
  { email: "user@example.com", username: "testuser" },
  { priority: 1, delay: 5000 }, // 可选配置
);

await Effect.runPromise(program);

5. 设置定时任务

参考 scheduled-job.md

使用 Job Schedulers API（BullMQ v5.16.0+）：

// 调度定时任务
const program = queueManager.scheduleJob(
  QueueName.CLEANUP,
  JobName.CLEANUP_DAILY,
  {},
  { pattern: "0 0 * * *" }, // Cron 表达式
);

await Effect.runPromise(program);

// 移除定时任务
await Effect.runPromise(
  queueManager.unscheduleJob(QueueName.CLEANUP, JobName.CLEANUP_DAILY),
);

模板参考

队列和任务定义

参考 queue-definition.md

包含：

• QueueName 枚举定义
• JobName 枚举定义
• 命名规范

Job Registry 类型映射

参考 job-definition.md

包含：

• Zod schema 定义（含验证规则）
• JobDefinitionRegistry 类型映射
• JobSchemaRegistry 验证映射
• QueueJobsMapping 队列关联

Worker 注册

参考 worker-registration.md

包含：

• Worker 注册位置（bootstrap.ts）
• Processor 函数实现
• Worker 配置选项
• 错误处理

定时任务设置

参考 scheduled-job.md

包含：

• Cron 表达式语法
• 调度和移除定时任务
• 查询已调度任务
• 定时任务最佳实践

完整示例

参考 examples/email-cleanup-queues.md 查看完整的 Email 和 Cleanup 队列实现。

关键 API

QueueManager 核心方法

// 添加任务（Effect 封装 + Zod 验证）
addJob<Q, N>(
  queueName: Q,
  jobName: N,
  data: JobDataByName<N>,
  opts?: JobsOptions,
): Effect<Job>

// 注册 Worker（Effect 封装）
registerWorker<Q>(
  queueName: Q,
  processor: (job: Job) => Promise<any>,
  options?: WorkerOptions,
): Effect<Worker>

// 调度定时任务（Effect 封装 + Zod 验证）
scheduleJob<Q, N>(
  queueName: Q,
  jobName: N,
  data: JobDataByName<N>,
  repeatOptions: RepeatOptions,
): Effect<void>

// 移除定时任务
unscheduleJob(queueName: string, jobName: string): Effect<boolean>

// 查询定时任务
getScheduledJobs(queueName: string): Effect<JobScheduler[]>

// 获取队列实例（用于 Bull Board）
getQueue(name: string): Queue

// 优雅关闭
close(timeoutMs?: number): Effect<void>

最佳实践

DO: 推荐做法

• 使用类型安全的 addJob 和 registerWorker 方法
• 所有 job data 定义 Zod schema 并添加中文错误消息
• Worker processor 中使用结构化日志（logger.info({ jobId, ... }, "[Queue]: message")）
• 定时任务使用明确的 cron 表达式（带注释）
• Worker 配置合理的并发数和重试策略
• 长时间运行的任务设置 timeout

DON'T: 避免做法

• 不要使用 getQueue 直接添加任务（绕过类型检查和验证）
• 不要在 job data 中传递大对象（使用引用 ID）
• 不要在 Worker 中使用 console.log（使用 logger）
• 不要忘记在 QueueJobsMapping 中关联队列和任务
• 不要修改已有 job 的 schema（创建新版本）
• 不要在 processor 中抛出未捕获异常（使用 try-catch）

集成点

与 Effect 系统集成

import { BullMQService } from "@/lib/infrastructure/effect/services/bullmq";

const program = Effect.gen(function* () {
  const qm = yield* BullMQService;
  yield* qm.addJob(QueueName.EMAIL, JobName.EMAIL_SEND_WELCOME, data);
});

与 db-schema 配合使用

定时任务可能需要查询数据库：

import { subDays } from "date-fns";
import { lt } from "drizzle-orm";
import db from "@/db";
import { auditLogs } from "@/db/schema";

// Worker processor 中
const processor = async (job: Job<CleanupData>) => {
  const { daysToKeep } = job.data;

  const cutoffDate = subDays(new Date(), daysToKeep);

  await db
    .delete(auditLogs)
    .where(lt(auditLogs.createdAt, cutoffDate.toISOString()));

  logger.info({ daysToKeep }, "[Cleanup]: 旧日志已清理");
};

Bull Board 监控

访问 /api/admin/queue-board 查看：

• 队列状态（waiting/active/completed/failed）
• 任务详情和重试
• Worker 性能指标
• 定时任务调度器

常见问题

Worker 何时注册？

在应用启动时（src/lib/infrastructure/bootstrap.ts）注册所有 Worker。

如何处理失败的任务？

BullMQ 自动重试（默认 3 次）。Worker 可配置：

registerWorker(queueName, processor, {
  attempts: 5,
  backoff: { type: "exponential", delay: 2000 },
});

定时任务如何避免重复？

使用 scheduleJob（内部调用 upsertJobScheduler），相同 schedulerId 会覆盖。

如何测试队列逻辑？

参考 src/lib/infrastructure/__tests__/bullmq-adapter.test.ts：

• 测试类型安全（编译时）
• 测试运行时验证（Zod）
• 测试 Worker processor 逻辑
• 测试定时任务调度

重要提醒

• BullMQ Worker 需要专用 Redis 连接（maxRetriesPerRequest: null）
• QueueManager 是 Singleton，自动在 shutdown 时销毁
• Worker 注册是幂等的（重复注册返回同一个实例）
• 定时任务使用 Job Schedulers API（v5.16.0+），不是 Repeatable Jobs
• Bull Board 动态获取队列（无需手动注册）