Fastify：如何在发送响应前等待Redis发布/订阅消息？

要实现Fastify路由等待Redis Pub/Sub消息后再响应，核心思路是将Redis的事件监听逻辑封装成Promise，这样就能用await阻塞路由执行，直到目标消息到达。同时要注意避免内存泄漏和请求超时问题，具体实现如下：

关键要点 #

• 为每个请求单独绑定一次性消息监听器，收到匹配消息后立即移除监听器
• 加入超时机制，防止请求因未收到消息长期挂起
• 确保Redis客户端已提前订阅目标频道（若未订阅，需先执行redis.subscribe('目标频道名')）

代码实现 #

1. 封装等待Redis消息的Promise函数 #

import Redis from 'ioredis';

function waitForRedisMessage(redis: Redis, targetId: string): Promise<any> {
  return new Promise((resolve, reject) => {
    // 设置10秒超时，可根据业务需求调整时长
    const timeoutTimer = setTimeout(() => {
      redis.off('message', messageHandler);
      reject(new Error('等待Redis消息超时'));
    }, 10000);

    function messageHandler(channel: string, message: string) {
      try {
        const msgData = JSON.parse(message);
        // 匹配当前请求的唯一id
        if (msgData.id === targetId) {
          clearTimeout(timeoutTimer);
          redis.off('message', messageHandler); // 移除监听器，避免内存泄漏
          resolve(msgData);
        }
      } catch (parseError) {
        clearTimeout(timeoutTimer);
        redis.off('message', messageHandler);
        reject(parseError);
      }
    }

    // 绑定消息监听器
    redis.on('message', messageHandler);
  });
}

2. 修改Fastify路由逻辑 #

app.post('/route', async function (req: any, rep) {
  try {
    // 等待匹配当前请求id的Redis消息
    await waitForRedisMessage(redis, req.body.id);
    rep.send('ok');
  } catch (error) {
    // 处理超时或消息解析错误
    rep.status(500).send(error.message);
  }
});

额外注意事项 #

• 分离Pub/Sub客户端：ioredis中处于订阅模式的客户端无法执行普通Redis命令（如get/set），建议单独创建订阅专用的Redis客户端，避免和业务命令客户端冲突。
• 频道过滤优化：如果Redis消息按频道分类发送，可以在messageHandler中先判断频道是否匹配，再处理消息内容，进一步缩小匹配范围。

内容的提问来源于stack exchange，提问作者Some Noob Student