如何在单个NestJS微服务中监听多个RabbitMQ队列？

针对你的需求，这里提供两种可行的实现方式：

方案一：使用queues数组配置（推荐） #

NestJS的RabbitMQ传输层支持通过queues数组一次性指定多个监听队列，替换原来的单个queue配置即可，代码修改如下：

const app = await NestFactory.createMicroservice<MicroserviceOptions>(
  AppModule,
  {
    transport: Transport.RMQ,
    options: {
      urls: [
        `amqp://${process.env.RABITMQ_USERNAME}:${process.env.RABITMQ_PASSWORD}@${process.env.RABITMQ_HOST}:${process.env.RABITMQ_PORT}`,
      ],
      // 替换单个queue为queues数组，包含两个目标队列
      queues: ['master_ms_locale', 'master_ms_country'],
      noAck: false,
      queueOptions: {
        durable: false,
      },
    },
  },
);

await app.listen();

配置完成后，需要在控制器中为每个队列编写对应的消息处理逻辑，示例如下：

import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';

@Controller()
export class AppController {
  // 处理master_ms_locale队列的请求消息（带回复）
  @MessagePattern('master_ms_locale')
  handleLocaleRequest(data: any) {
    console.log('收到locale队列消息:', data);
    // 这里编写你的业务处理逻辑
    return 'locale消息处理完成';
  }

  // 处理master_ms_country队列的请求消息（带回复）
  @MessagePattern('master_ms_country')
  handleCountryRequest(data: any) {
    console.log('收到country队列消息:', data);
    // 这里编写你的业务处理逻辑
    return 'country消息处理完成';
  }
}

如果你的队列是用于事件通知（无需回复消息），可以把@MessagePattern换成@EventPattern，处理函数不需要返回值。

方案二：创建多个微服务实例 #

如果需要为不同队列配置不同的RabbitMQ参数（比如不同的noAck、queueOptions），可以创建两个独立的微服务实例，分别监听对应队列：

// 实例1：监听master_ms_locale队列
const localeMicroservice = await NestFactory.createMicroservice<MicroserviceOptions>(
  AppModule,
  {
    transport: Transport.RMQ,
    options: {
      urls: [
        `amqp://${process.env.RABITMQ_USERNAME}:${process.env.RABITMQ_PASSWORD}@${process.env.RABITMQ_HOST}:${process.env.RABITMQ_PORT}`,
      ],
      queue: 'master_ms_locale',
      noAck: false,
      queueOptions: {
        durable: false,
      },
    },
  },
);

// 实例2：监听master_ms_country队列
const countryMicroservice = await NestFactory.createMicroservice<MicroserviceOptions>(
  AppModule,
  {
    transport: Transport.RMQ,
    options: {
      urls: [
        `amqp://${process.env.RABITMQ_USERNAME}:${process.env.RABITMQ_PASSWORD}@${process.env.RABITMQ_HOST}:${process.env.RABITMQ_PORT}`,
      ],
      queue: 'master_ms_country',
      noAck: false,
      queueOptions: {
        durable: false,
      },
    },
  },
);

// 同时启动两个微服务
await Promise.all([localeMicroservice.listen(), countryMicroservice.listen()]);

这种方式灵活性更高，每个队列可以单独配置，但代码量会稍多一些。

内容的提问来源于stack exchange，提问作者Dilpesh