如何使用NestJS Golevelup RabbitMQ的RabbitSubscribe单条消费消息？

要实现每次仅消费一条消息（上一条处理完成后才接收下一条），需确保两个核心配置：通道预取计数设为1 + 开启手动消息确认，以下是具体实现步骤：

1. 配置带预取限制的RabbitMQ通道 #

在模块的RabbitMQ初始化配置中，为指定通道设置prefetchCount: 1，限制该通道最多预取1条未确认的消息：

import { Module } from '@nestjs/common';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [
        {
          name: 'test_exchange',
          type: 'direct',
        },
      ],
      channels: {
        'channel-2': {
          prefetchCount: 1, // 关键：限制通道仅预取1条未确认消息
          default: false,
        },
      },
      uri: 'amqp://localhost:5672', // 替换为你的RabbitMQ服务地址
    }),
  ],
})
export class AppModule {}

2. 调整消费者配置，开启手动确认 #

修改@RabbitSubscribe装饰器，关闭自动确认并在消息处理完成后手动确认：

import { RabbitSubscribe, RabbitMessage } from '@golevelup/nestjs-rabbitmq';

// ...

@RabbitSubscribe({
  exchange: 'test_exchange',
  routingKey: 'test_update',
  queue: 'test_update',
  queueOptions: {
    channel: 'channel-2', // 使用配置好的带预取限制的通道
  },
  noAck: false, // 必须关闭自动确认，开启手动模式
})
public async updateItem(msg: any, amqpMsg: RabbitMessage) {
  try {
    await this.service.updateItem(msg);
    // 处理成功后手动确认，RabbitMQ才会推送下一条消息
    amqpMsg.ack();
  } catch (error) {
    // 处理失败时可拒绝消息，第二个参数为true则消息重新入队，false直接丢弃
    amqpMsg.nack(false, false);
  }
}

关键注意事项 #

• 若保持默认自动确认（noAck: true），即使设置了预取计数，RabbitMQ也会在消息被接收后立即标记为已处理，继续推送新消息，必须关闭自动确认才能生效。
• 确保prefetchCount配置在消费者指定的通道上，否则限制不会生效。

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