Pub/Sub、阻塞命令与连接池

这些能力共同涉及一个关键概念：有些操作会长期占用连接，因此不能随意与普通命令共用。

Pub/Sub 是什么

发布者向频道发送消息，当前订阅者实时收到消息：

publisher -> channel -> subscriber A
                     -> subscriber B

Redis Pub/Sub 不保存历史消息。订阅者离线时发送的消息不会在重新上线后补发。

使用独立订阅连接

RESP2 模式下，进入订阅状态的连接不能再执行普通命令。因此官方建议复制客户端：

const subscriber = client.duplicate();

subscriber.on("error", (error: Error) => {
  console.error("Redis subscriber error", error);
});

await subscriber.connect();

const listener = (message: string, channel: string): void => {
  console.log(`频道 ${channel} 收到：${message}`);
};

await subscriber.subscribe("nloop:notifications", listener);

发布使用普通客户端：

await client.publish(
  "nloop:notifications",
  JSON.stringify({ type: "USER_CREATED", userId: 42 }),
);

取消订阅并关闭：

await subscriber.unsubscribe("nloop:notifications", listener);
await subscriber.close();

重复订阅同一频道会注册多个监听器，每个监听器都会收到消息。

什么时候使用 Pub/Sub

适合：

不适合：

这些场景应考虑 Redis Streams 或专业消息队列。

阻塞命令

BLPOP 会等待列表出现元素：

const blockingClient = client.duplicate();
blockingClient.on("error", console.error);
await blockingClient.connect();

const item = await blockingClient.blPop("nloop:jobs", 10);

等待期间这条连接不能正常承担其他请求。不要在应用共享的主客户端上执行长时间阻塞命令。

RedisClientPool

Node-Redis 新版将隔离连接池提取为独立的 RedisClientPool，适合需要多条独占连接的场景：

import { createClientPool } from "redis";

const pool = await createClientPool()
  .on("error", (error: Error) => {
    console.error("Redis pool error", error);
  })
  .connect();

await pool.ping();

普通 GET/SET 通常只需要一个共享客户端，不要照搬数据库连接池的经验给 Redis 建几十条连接。

消息格式最佳实践

定义清晰的消息结构：

interface UserCreatedEvent {
  version: 1;
  type: "USER_CREATED";
  occurredAt: string;
  data: {
    userId: number;
  };
}