Nestjs教程消息队列集成

如何在NestJS中集成消息队列功能？我想在项目中实现异步任务处理，比如发送邮件或处理耗时操作，但不清楚具体该选择RabbitMQ还是Redis作为消息队列。能否提供从安装配置到实际应用的完整教程？特别需要了解：

1)如何安装和配置消息队列中间件； 2)在NestJS中创建生产者和消费者的最佳实践； 3)如何处理消息失败的重试机制； 4)如何监控队列状态。

2 回复

可以学学这个：https://www.itying.com/goods-1199.html

要在NestJS中集成消息队列，常用的是RabbitMQ或Kafka。以下是基于RabbitMQ的简单教程：

  1. 安装依赖：
npm install @golevelup/nestjs-rabbitmq amqplib
  1. 配置模块： 创建一个rabbitmq.module.ts，使用@golevelup/nestjs-rabbitmq提供的装饰器配置连接：
import { Module } from '@nestjs/common';
import { RabbitMqModule } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMqModule.forRootAsync({
      useFactory: () => ({
        uri: 'amqp://localhost',
        noAck: false,
        queue: 'test_queue',
      }),
    }),
  ],
})
export class AppModule {}
  1. 创建生产者： 在服务中注入RabbitMqService来发送消息：
import { Injectable } from '@nestjs/common';
import { RabbitMqService } from '@golevelup/nestjs-rabbitmq';

@Injectable()
export class AppService {
  constructor(private rabbitMq: RabbitMqService) {}

  async sendMessage(message: string) {
    await this.rabbitMq.sendJson('test_queue', { message });
  }
}
  1. 创建消费者： 监听队列中的消息：
import { Message, RabbitMqController } from '@golevelup/nestjs-rabbitmq';

@Controller('test_queue')
export class TestQueueController {
  @Message('test_queue')
  handleMessage(data: any) {
    console.log('Received:', data);
  }
}

完成以上步骤后，你的NestJS应用就可以通过RabbitMQ实现消息的发送和接收了。


NestJS 消息队列集成教程

NestJS 提供了与几种流行消息队列系统的集成方式，包括 RabbitMQ、Kafka 等。以下是主要消息队列的集成方法：

1. RabbitMQ 集成

首先安装所需依赖：

npm install --save @nestjs/microservices amqplib amqp-connection-manager

基本配置

// main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.RMQ,
      options: {
        urls: ['amqp://localhost:5672'],
        queue: 'cats_queue',
        queueOptions: {
          durable: false
        },
      },
    },
  );
  await app.listen();
}
bootstrap();

消息处理器

// cats.controller.ts
import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class CatsController {
  @MessagePattern('cats')
  getNotifications(@Payload() data: any) {
    console.log(data);
    return { response: 'Received' };
  }
}

2. Kafka 集成

安装依赖：

npm install --save @nestjs/microservices kafkajs

基本配置

// main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.KAFKA,
      options: {
        client: {
          brokers: ['localhost:9092'],
        },
        consumer: {
          groupId: 'cats-consumer',
        },
      },
    },
  );
  await app.listen();
}
bootstrap();

消息处理器

// cats.controller.ts
import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class CatsController {
  @MessagePattern('cats')
  getNotifications(@Payload() data: any) {
    console.log(data);
    return { response: 'Received' };
  }
}

注意事项

  1. 确保消息队列服务（RabbitMQ/Kafka）已安装并运行
  2. 根据生产环境需求配置队列持久化、消费者组等参数
  3. 在处理消息时考虑幂等性和错误处理机制
  4. 对于高吞吐量场景，可能需要调整预取计数(prefetch count)等参数

NestJS 的消息队列集成提供了强大的抽象，使开发者可以专注于业务逻辑而非基础设施细节。

回到顶部