1.安裝rabbitmq: docker run -dit --name myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management

2.npm install amqplib amqp-connection-manager


3.subscribers.module.ts

import { Module } from '@nestjs/common';
import { SubscribersController } from './subscribers.controller';
import { SubscribeService } from 'src/modules/subscribers.service'
import { ConfigModule, ConfigService } from '@nestjs/config';
import { ClientProxyFactory, Transport } from '@nestjs/microservices';

@Module({
    imports: [ConfigModule],
    controllers: [SubscribersController],
    providers: [
        {
            provide: 'SUBSCRIBERS_SERVICE',
            useFactory: () => {
                return ClientProxyFactory.create({
                    transport: Transport.RMQ,
                    options: {
                        urls: [`amqp://admin:admin@localhost:5672`],
                        queue: 'my_queue2',
                        noAck: false,//false,需要手動確認消息
                        queueOptions: {
                            durable: true,//消息持久化
                        },
                    },
                })
            },
            inject: [ConfigService],
        }, SubscribeService
    ],
})
export class SubscribersModule { }

4.subscribers.controller.ts//發送消息

import {
    Body,
    Controller,
    Post,
    Inject,
} from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { SubscribeService } from 'src/modules/subscribers.service'
@Controller('subscribers')
export class SubscribersController {
    constructor(
        @Inject('SUBSCRIBERS_SERVICE') private subscribersService: ClientProxy,

        private service: SubscribeService
    ) { }
    @Post()
    async createPost(@Body() subscriber) {
        // this.service.createPost(subscriber)
        //省略數據庫存儲,
        return this.subscribersService.send({
            cmd: 'add-subscriber'
        }, subscriber)

    }

    //省略定時任務重新發送
}

5.app.controller.ts //接受消息

import { Controller, Get, Inject, Body, Post } from '@nestjs/common';
import { AppService } from './app.service';
import { MessagePattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';
@Controller()
export class AppController {
  constructor(
    private readonly appService: AppService,


  ) { }

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @MessagePattern({ cmd: 'add-subscriber' })
  async addSubscriber(@Payload() subscriber, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const originalMsg = context.getMessage();
    //省略處理數據庫
    console.log('app.....get message is ', subscriber)
    channel.ack(originalMsg);
    return subscriber;
  }
}

6.main.ts

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { Transport } from "@nestjs/microservices";
async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.connectMicroservice({
    transport: Transport.RMQ,
    options: {
      urls: [`amqp://admin:admin@localhost:5672`],
      queue: 'my_queue2',
      noAck: false,
      queueOptions: { durable: true },
    },
  });
  app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();