Nestjs 微服务的使用

官方文档：https://docs.nestjs.com/microservices/basics

官方示列demo：https://github.com/nestjs/nest/tree/master/sample/03-microservices

1、安装微服务

npm i --save @nestjs/microservices

2、main.ts进行配置

import { NestFactory } from '@nestjs/core';
import { Transport } from '@nestjs/microservices';

import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.connectMicroservice({
    transport: Transport.TCP,
  });

  await app.startAllMicroservicesAsync();
  await app.listen(5000);
}
bootstrap();

3、app.controller.ts配置

import { Controller, Get, Query, UsePipes } from '@nestjs/common';
import {
  MessagePattern,
  Client,
  Transport,
  ClientProxy,
} from '@nestjs/microservices';
import { AppService } from './app.service';

import { Observable, from } from 'rxjs';

import { ValidationPipe } from './validation.pipe';

@Controller()
export class AppController {
  @Client({ transport: Transport.TCP })
  client: ClientProxy;

  constructor(private readonly appService: AppService) {}

  @UsePipes(new ValidationPipe())
  @Get()
  getHello(@Query('data') data): Observable<number> {
    // const pattern = { cmd: 'sumObservable' };
    // const pattern = { cmd: 'sumAsync' };
    const pattern = { cmd: 'sum' };

    // 使用 send 调用微服务
    const r = this.client.send<number>(pattern, data);
    return r;
  }

  @MessagePattern({ cmd: 'sum' })
  sum(data: number[]): number {
    return data.reduce((acc, el) => acc + el);
  }

  // 返回promise异步响应
  @MessagePattern({ cmd: 'sumAsync' })
  sumAsync(data: number[]): Promise<number> {
    const result = data.reduce((acc, el) => acc + el) + 1;
    return Promise.resolve(result);
  }

  // 程序将响应3次
  @MessagePattern({ cmd: 'sumObservable' })
  sumObservable(data: number[]): Observable<number> {
    return from([1, 2, 3]);
  }
}

4、validation.pipe.ts

import {
  PipeTransform,
  Injectable,
  ArgumentMetadata,
  PayloadTooLargeException,
  BadRequestException,
} from '@nestjs/common';

@Injectable()
export class ValidationPipe implements PipeTransform {
  transform(value: any, metadata: ArgumentMetadata) {
    if (metadata.type === 'query') {
      try {
        return JSON.parse(value);
      } catch (error) {
        throw new BadRequestException();
      }
    } else {
      throw new PayloadTooLargeException();
    }
  }
}

访问： http://localhost:5000/?data=[1,2, 3] // 6