03 - Redis 连接管理

BullMQ 的 Queue、Worker、QueueEvents 等对象都需要访问后端。默认情况下，BullMQ 使用 ioredis 建立 Redis 连接；也支持通过适配器使用 node-redis 等客户端。

1. 最简单的连接配置

把连接选项交给 BullMQ：

import { Queue, Worker } from "bullmq";

interface ResizeJobData {
  imageId: string;
}

const connection = {
  host: "127.0.0.1",
  port: 6379,
};

const queue = new Queue<ResizeJobData>("image-resize", { connection });

const worker = new Worker<ResizeJobData>(
  "image-resize",
  async (job): Promise<void> => {
    console.log(`处理图片 ${job.data.imageId}`);
  },
  { connection },
);

这里复用的是普通配置对象，不是同一个网络连接。每个 BullMQ 实例会按需要创建连接。

未提供连接选项时，默认地址是 localhost:6379。示例中仍显式写出连接，便于初学者看懂依赖关系。

2. 复用 ioredis 客户端

当 Redis 服务商限制连接数时，可以手动创建客户端：

import { Queue } from "bullmq";
import IORedis from "ioredis";

const producerConnection = new IORedis({
  host: "127.0.0.1",
  port: 6379,
  maxRetriesPerRequest: 1,
});

const emailQueue = new Queue("email", {
  connection: producerConnection,
});

const reportQueue = new Queue("report", {
  connection: producerConnection,
});

多个普通 Queue 可以共享一个客户端。不要为了“连接越少越好”过度优化；官方文档指出 Redis 连接的开销通常较低，只有存在连接数限制或明确运维需求时才值得集中复用。

3. Worker 为什么仍会增加连接

Worker 需要使用阻塞式 Redis 命令等待新任务。阻塞连接不能同时承担普通命令，因此即使把已有客户端传给 Worker，Worker 仍会在内部调用 duplicate() 创建专用阻塞连接。

import { Worker } from "bullmq";
import IORedis from "ioredis";

const workerConnection = new IORedis({
  host: "127.0.0.1",
  port: 6379,
  maxRetriesPerRequest: null,
});

const worker = new Worker(
  "email",
  async (job): Promise<void> => {
    console.log(job.data);
  },
  { connection: workerConnection },
);

同样需要阻塞命令的 QueueEvents 也会使用额外连接。规划连接数量时不能简单地按“创建了几个客户端对象”计算。

4. maxRetriesPerRequest 的关键区别

该 ioredis 选项控制一条命令失败后重试多少次才向调用者报错。

HTTP 生产者：应快速失败

用户请求不能因为 Redis 下线而无限等待。生产者通常保留有限重试，或设为较小值：

const producerConnection = new IORedis({
  host: "127.0.0.1",
  port: 6379,
  maxRetriesPerRequest: 1,
});

入队失败后，接口应返回明确错误，让调用方稍后重试，而不是错误地返回“任务已创建”。

Worker：应持续等待 Redis 恢复

后台 Worker 通常需要在 Redis 暂时不可用时持续重连：

const workerConnection = new IORedis({
  host: "127.0.0.1",
  port: 6379,
  maxRetriesPerRequest: null,
});

官方文档说明：如果手动创建 ioredis 客户端并传给 Worker，BullMQ 要求 maxRetriesPerRequest 为 null。

因此，HTTP 生产者和 Worker 通常不应共享同一个 ioredis 实例，它们的可用性目标不同。

5. 使用 node-redis 适配器

新版 BullMQ 官方文档也提供 node-redis 适配方式。BullMQ 不会替你创建 node-redis 客户端，需要先创建原始客户端，再包装：

import { Queue, Worker, createNodeRedisClient } from "bullmq";
import { createClient } from "redis";

interface EmailJobData {
  to: string;
}

const rawClient = createClient({
  url: "redis://127.0.0.1:6379",
});

const connection = createNodeRedisClient(rawClient);

const queue = new Queue<EmailJobData>("email", { connection });
const worker = new Worker<EmailJobData>(
  "email",
  async (job): Promise<void> => {
    console.log(job.data.to);
  },
  { connection },
);

使用该适配器时，官方要求安装 redis 5 或更高版本。初学阶段建议先使用官方示例最常见的默认 ioredis 路径，理解 Queue 和 Worker 后再比较客户端差异。

6. 使用环境变量

不要把生产密码写进源码：

interface RedisConnectionConfig {
  host: string;
  port: number;
  password?: string;
}

function getRedisConnection(): RedisConnectionConfig {
  const port = Number(process.env.REDIS_PORT ?? "6379");

  if (!Number.isInteger(port) || port <= 0) {
    throw new Error("REDIS_PORT 必须是有效端口");
  }

  const password = process.env.REDIS_PASSWORD;

  return {
    host: process.env.REDIS_HOST ?? "127.0.0.1",
    port,
    ...(password === undefined ? {} : { password }),
  };
}

这里用条件展开避免在启用 exactOptionalPropertyTypes 时把 undefined 显式赋给可选字段。

7. 正确关闭连接

关闭顺序应是：先停止使用连接的 BullMQ 对象，再关闭共享连接。

async function shutdown(): Promise<void> {
  await worker.close();
  await queue.close();
  await workerConnection.quit();
  await producerConnection.quit();
}

如果连接由 BullMQ 根据配置对象自行创建，调用对应对象的 close() 即可。若是自己创建并共享的客户端，则在所有使用者关闭之后，再由应用负责 quit() 或 disconnect()。

不要在每次入队后都创建并关闭 Redis 连接。Web 服务应在启动时创建 Queue，在进程退出时统一关闭。

8. 生产环境注意事项

本章小结

Queue 更关心快速回应调用方，Worker 更关心在故障后继续恢复，因此二者的连接重试策略不同。连接可以复用，但 Worker 和 QueueEvents 因阻塞式等待仍会创建额外连接；关闭时必须先关 BullMQ 实例，再关共享客户端。

官方资料