使用@golevelup/nestjs-rabbitmq遇AMQP连接不可用问题求助

问题描述 #

使用@golevelup/nestjs-rabbitmq时，配置连接管理器为不等待连接，已编写AppModule、RabbitMQ模块、配置文件、服务及控制器相关代码。访问接口http://localhost:3030/api/v1/amqp/sending时，抛出「AMQP connection is not available」错误。

代码示例 #

app.module.ts #

...
import RabbitModule from '@/modules/rabbitmq/rabbitmq.module'
@Module({
  imports: [..., RabbitModule],
  controllers: [],
  providers: []
})
export default class AppModule {}

modules\rabbitmq\rabbitmq.module.ts #

...
@Module({
  imports: [RabbitMQConf, RabbitMQMessagingModule],
  providers: [RabbitmqService],
  controllers: [RabbitmqController],
  exports: [RabbitmqService]
})
export default class RabbitMQMessagingModule {}

configs\rabbitmq.ts #

...
@Module({
  imports: [RabbitMQConf, RabbitMQMessagingModule],
  providers: [RabbitmqService],
  controllers: [RabbitmqController],
  exports: [RabbitmqService]
})
export default class RabbitMQMessagingModule {}

modules\rabbitmq\rabbitmq.service.ts #

...
@Injectable()
export default class RabbitmqService {
  constructor(private readonly amqpConnection: AmqpConnection) {}
...
  public publish(): void {
    this.amqpConnection.publish('amq.fanout', 'amq_fanout_queue_iot', { msg: 'hello world' })
  }
}

modules\rabbitmq\rabbitmq.controller.ts #

import { Controller, Get } from '@nestjs/common'
import Unbanning from '@/decorators/unbanning.decorator'
import RabbitmqService from './rabbitmq.service'
@Controller('amqp')
export default class RabbitmqController {
  constructor(private readonly amqp: RabbitmqService) {}
  @Unbanning()
  @Get('sending')
  send() {
    return this.amqp.publish()
  }
}

错误信息 #

[ exception ]-26 Error: AMQP connection is not available
    at AmqpConnection.publish (F:\nestjs\node_modules\.pnpm\@golevelup+nestjs-rabbitmq@3.6.1_@nestjs+common@9.4.2_rxjs@7.8.1\node_modules\@golevelup\nestjs-rabbitmq\src\amqp\connection.ts:544:13)
    at RabbitmqService.publish (F:\nestjs\src\modules\rabbitmq\rabbitmq.service.ts:58:25)
    at RabbitmqController.send (F:\nestjs\src\modules\rabbitmq\rabbitmq.controller.ts:22:22)
    at 
...

解决方法 #

• 确认waitForConnection配置
  确保在RabbitMQ连接配置中明确设置waitForConnection: false，注意该配置仅让模块启动时不阻塞等待连接，但调用publish前仍需保证连接已建立。

• 等待连接就绪后再发布
  修改publish方法，通过init()等待连接完成，同时控制器改为异步调用：

  // rabbitmq.service.ts
  public async publish(): Promise<void> {
    // 等待连接初始化完成
    await this.amqpConnection.init();
    this.amqpConnection.publish('amq.fanout', 'amq_fanout_queue_iot', { msg: 'hello world' });
  }
  

  // rabbitmq.controller.ts
  @Unbanning()
  @Get('sending')
  async send() {
    await this.amqp.publish();
    return { status: 'message sent' };
  }
  

• 修复模块导入循环
  当前配置模块与业务模块存在循环导入问题，重构配置模块：

  // configs\rabbitmq.ts
  import { Module } from '@nestjs/common';
  import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
  import { ConfigService } from '@nestjs/config';
  
  @Module({
    imports: [
      RabbitMQModule.forRootAsync(RabbitMQModule, {
        useFactory: (config: ConfigService) => ({
          uri: config.get('RABBITMQ_URI'),
          exchanges: [{ name: 'amq.fanout', type: 'fanout' }],
          waitForConnection: false,
        }),
        inject: [ConfigService],
      }),
    ],
    exports: [RabbitMQModule],
  })
  export class RabbitMQConf {}
  

  业务模块仅导入配置模块：

  // modules\rabbitmq\rabbitmq.module.ts
  @Module({
    imports: [RabbitMQConf],
    providers: [RabbitmqService],
    controllers: [RabbitmqController],
    exports: [RabbitmqService],
  })
  export default class RabbitMQMessagingModule {}
  

• 检查RabbitMQ服务状态
  确认RabbitMQ服务正常运行，连接地址、端口、用户名密码等配置与实际一致。可通过rabbitmqctl status命令或管理后台（默认http://localhost:15672）验证服务可用性。

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