02 - 安装与快速开始

本章用一个最小示例走通 BullMQ 的完整链路：生产者把任务写入 Redis，Worker 从 Redis 取出任务并处理。

1. 开始前需要什么

BullMQ 是 Node.js 的任务队列库，默认以 Redis 作为队列后端。因此至少需要：

• Node.js 与 npm；
• 一个正在运行的 Redis 实例；
• BullMQ 包；
• TypeScript 与 ts-node（本项目已经具备）。

BullMQ 不是 Redis 服务器。安装 bullmq 以后，仍然必须单独启动 Redis。

用 Docker 启动本地 Redis

如果本机已经安装 Docker，可以执行：

docker run --name bullmq-redis -p 6379:6379 -d redis:7-alpine

确认 Redis 可用：

docker exec bullmq-redis redis-cli ping

返回 PONG 表示连接正常。

生产环境不要直接照搬这个无密码、无持久化配置。生产环境还要考虑认证、TLS、持久化、备份与 maxmemory-policy=noeviction。

2. 安装 BullMQ

官方快速开始使用：

npm install bullmq

如果后续要手动创建并复用 ioredis 连接，建议把直接使用的依赖也显式安装：

npm install bullmq ioredis

这样做能清楚表达项目直接依赖了 ioredis，而不是依赖包管理器恰好把 BullMQ 的传递依赖提升到了顶层。

本教程只展示命令，不会替你修改当前项目的 package.json。

3. 最小生产者

创建 producer.ts：

import { Queue } from "bullmq";

interface EmailJobData {
  to: string;
  subject: string;
}

const queueName = "email";

const emailQueue = new Queue<EmailJobData>(queueName, {
  connection: {
    host: "127.0.0.1",
    port: 6379,
  },
});

async function main(): Promise<void> {
  const job = await emailQueue.add("send-welcome-email", {
    to: "student@example.com",
    subject: "欢迎学习 BullMQ",
  });

  console.log(`任务已进入队列，jobId=${job.id}`);
}

main()
  .catch((error: unknown) => {
    console.error("添加任务失败", error);
    process.exitCode = 1;
  })
  .finally(async () => {
    await emailQueue.close();
  });

运行：

ts-node producer.ts

即使 Worker 还没有启动，任务也可以先进入 Redis，等待稍后消费。

4. 最小 Worker

创建 worker.ts：

import { Job, Worker } from "bullmq";

interface EmailJobData {
  to: string;
  subject: string;
}

interface EmailJobResult {
  sentAt: string;
}

const queueName = "email";

const worker = new Worker<EmailJobData, EmailJobResult>(
  queueName,
  async (job: Job<EmailJobData>): Promise<EmailJobResult> => {
    console.log(`正在处理任务 ${job.id}`, job.data);

    // 这里用延时模拟调用邮件服务。
    await new Promise<void>((resolve) => setTimeout(resolve, 500));

    return { sentAt: new Date().toISOString() };
  },
  {
    connection: {
      host: "127.0.0.1",
      port: 6379,
    },
  },
);

worker.on("completed", (job, result) => {
  console.log(`任务 ${job.id} 完成`, result);
});

worker.on("failed", (job, error: Error) => {
  console.error(`任务 ${job?.id ?? "unknown"} 失败`, error);
});

worker.on("error", (error: Error) => {
  // 官方文档特别提醒：Worker 应监听 error 事件。
  console.error("Worker 连接或内部错误", error);
});

在另一个终端运行：

ts-node worker.ts

再次运行生产者，就能看到 Worker 处理任务。生产者和 Worker 的队列名必须完全一致。

5. 当前项目为什么能使用 import

当前项目的 tsconfig.json 使用：

{
  "compilerOptions": {
    "module": "commonjs"
  }
}

源码仍然可以写 import { Queue } from "bullmq"。ts-node 在运行时会按照 TypeScript 配置，将 import 转换成 CommonJS 的加载形式，因此可以直接执行：

ts-node producer.ts
ts-node worker.ts

这并不代表项目已经切换成原生 ESM。若将来改为真正的 ESM，还需要同步调整 package.json 的 "type"、tsconfig.json 的模块设置以及启动命令。

6. 一次请求应该等待任务完成吗

通常不应该。Web 接口只负责把任务加入队列，然后尽快返回：

const job = await emailQueue.add("send-welcome-email", data);

response.status(202).json({
  message: "任务已受理",
  jobId: job.id,
});

202 Accepted 表示任务已经被接受，但尚不保证处理完成。客户端可使用 jobId 查询状态，或由服务端通过 WebSocket、回调等方式通知结果。

7. 初学者常见问题

Worker 一启动就开始消费吗

是。官方文档说明，创建 Worker 后默认立即运行。若要先完成其他初始化，可以设置 autorun: false，之后手动调用 worker.run()。

需要自己轮询 Redis 吗

不需要。BullMQ 负责等待任务、锁、状态迁移和失败处理。业务代码只需要实现 processor 函数。

还需要 QueueScheduler 吗

BullMQ 2.0 及之后，延迟任务和 stalled 恢复不再要求额外启动 QueueScheduler。看到旧教程时要留意其适用版本。

实用技巧与最佳实践

• Queue 与 Worker 使用集中定义的队列名，避免拼写不一致。
• 生产者只放任务所需的最小数据，不要把请求对象、数据库连接等放进任务。
• Worker 必须监听 error 事件，并记录 failed 事件。
• 将 Web 服务与 Worker 作为两个独立进程运行，便于独立扩容和重启。
• Worker 执行逻辑应尽量幂等，因为极端情况下任务可能被再次处理。
• 本地学习可连接 127.0.0.1:6379；实际项目应通过环境变量提供连接信息。

本章小结

BullMQ 的最小系统只有三个角色：Queue 是生产者入口，Redis 保存队列状态，Worker 执行任务。下一章将深入解释连接为什么不能一概复用，以及生产者和 Worker 为什么需要不同的失败策略。

官方资料

• Quick Start
• Queues
• Workers