如果您使用@golevelup/nestjs-rabbitmq包，您将受益匪浅，因为它支持不同的消息队列消费，如果您的应用是混合型的，还可以支持更多。

npm i @golevelup/nestjs-rabbitmq

那么nestjs应用程序结构应该如下所示

src --
     |
     app.module.ts
     main.ts
     app.module.ts
     someHttpModule1 --
                      |
                      someHttpModule1.controller.ts
                      someHttpModule1.module.ts
                      someHttpModule1.service.ts
                      ...
     someHttpModule2 --
                      |
                      someHttpModule2.controller.ts
                      someHttpModule2.module.ts
                      someHttpModule2.service.ts
                      ...
     ...
     // Events module is designed for consuming messages from rabbitmq
     events --
             |
             events.module.ts
             someEventConsumerModule1 --
                                       |
                                       someEventConsumerModule1.module.ts
                                       someEventConsumerModule1.service.ts
             someEventConsumerModule2 --
                                       |
                                       someEventConsumerModule2.module.ts
                                       someEventConsumerModule2.service.ts
             ...

在src/app.module.ts文件中

// module imports
import { SomeHttpModule1 } from './someHttpModule1/someHttpModule1.module'
import { SomeHttpModule2 } from './someHttpModule2/someHttpModule.module'
import { EventsModule } from './events/events.module'
// and other necessery modules

@Module(
  imports: [
    SomeHttpModule1, 
    SomeHttpModule2,
    EventsModule, 
    // and other dependent modules
  ],
  controller: [],
  providers: []
})

export class AppModule{}

并且在事件.module.ts文件中

// imports
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq'
import { SomeEventConsumerModule1 } from './someEventConsumerModule1/someEventConsumerModule1.module'
import { SomeEventConsumerModule2 } from './someEventConsumerModule2/someEventConsumerModule2.module'
// and so on

@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [{
        name: 'amq.direct',
        type: 'direct' // check out docs for more information on exchange types
      }],
      uri: 'amqp://guest:guest@localhost:5672', // default login and password is guest, and listens locally to 5672 port in amqp protocol
      connectionInitOptions: { wait: false }
    }),
    SomeEventConsumerModule1,
    SomeEventConsumerModule2,
    // ... and other dependent consumer modules 
  ]
})
export class EventsModule {}

下面是一个使用者模块的示例（someEventConsumerModule1.module.ts）

// imports
import { SomeEventConsumerModule1Service } from './someEventConsumerModule1.service'
// ...

@Module({
  imports: [
    SomeEventConsumerModule1,
    // other modules if injected
  ],
  providers: [
    SomeEventConsumerModule1Service
  ]
})
export class SomeEventConsumerModule1 {}

并在服务文件中放置业务逻辑如何处理消息

// imports
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq'
import { Injectable } from '@nestjs/common'
import { ConsumeMessage, Channel } from 'amqplib' // for type safety you will need to install package first
// ... so on

@Injectable()
export class SomeEventConsumerModule1Service {
  constructor(
    // other module services if needs to be injected
  ) {}
  @RabbitSubscribe({
    exchange: 'amq.direct',
    routingKey: 'direct-route-key', // up to you
    queue: 'queueNameToBeConsumed',
    errorHandler: (channel: Channel, msg: ConsumeMessage, error: Error) => {
      console.log(error)
      channel.reject(msg, false) // use error handler, or otherwise app will crush in not intended way
    }
  })
  public async onQueueConsumption(msg: {}, amqpMsg: ConsumeMessage) {
    const eventData = JSON.parse(amqpMsg.content.toString())
    // do something with eventData
    console.log(`EventData: ${eventData}, successfully consumed!`)
  }
  // ... and in the same way
}