Tổng quan (Overview)

Bên cạnh kiến trúc ứng dụng truyền thống (đôi khi gọi là monolithic), Nest hỗ trợ sẵn kiểu kiến trúc microservice. Hầu hết các khái niệm đã nói ở những phần khác trong tài liệu này — như dependency injection, decorator, exception filter, pipe, guard và interceptor — đều áp dụng tương tự cho microservices. Ở bất cứ đâu có thể, Nest trừu tượng hóa chi tiết triển khai để cùng một component có thể chạy trên các nền tảng dựa trên HTTP, WebSockets và Microservices. Phần này nói về những khía cạnh của Nest dành riêng cho microservices.

Trong Nest, một microservice về bản chất là một ứng dụng sử dụng lớp transport khác với HTTP.

Nest hỗ trợ sẵn nhiều cách triển khai lớp transport, gọi là transporter, có nhiệm vụ truyền message giữa các instance microservice khác nhau. Hầu hết các transporter hỗ trợ sẵn cả hai kiểu message: request-response và event-based. Nest trừu tượng hóa chi tiết triển khai của từng transporter phía sau một interface chung cho cả messaging kiểu request-response và event-based. Nhờ vậy bạn dễ dàng chuyển từ transport layer này sang transport layer khác — ví dụ để tận dụng tính năng độ tin cậy hay hiệu năng riêng của một transport layer cụ thể — mà không ảnh hưởng đến code ứng dụng.

Cài đặt (Installation)

Để bắt đầu xây dựng microservices, trước tiên cài package cần thiết:

$ npm i --save @nestjs/microservices

Bắt đầu (Getting started)

Để khởi tạo một microservice, dùng method createMicroservice() của class NestFactory:

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

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.TCP,
    },
  );
  await app.listen();
}
await bootstrap();

import { NestFactory } from '@nestjs/core';
import { Transport } from '@nestjs/microservices';
import { AppModule } from './app.module.js';

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

Tham số thứ hai của method createMicroservice() là một object options. Object này có thể gồm hai thành phần:

Object options phụ thuộc vào transporter bạn chọn. Transporter TCP có các thuộc tính được mô tả dưới đây. Với các transporter khác (ví dụ Redis, MQTT, v.v.), xem chương tương ứng để biết các option khả dụng.

Message pattern và Event pattern (Message and Event Patterns)

Microservices nhận diện cả message và event thông qua pattern. Pattern là một giá trị đơn giản, ví dụ một object literal hoặc một string. Pattern được tự động serialize và gửi qua network kèm phần data của message. Nhờ vậy, bên gửi và bên nhận message có thể phối hợp xác định request nào được xử lý bởi handler nào.

Request-response

Kiểu message request-response hữu ích khi bạn cần trao đổi message giữa các service bên ngoài với nhau. Mô hình này đảm bảo rằng service đã thực sự nhận được message (mà bạn không cần tự triển khai một cơ chế acknowledgment). Tuy nhiên, cách tiếp cận request-response không phải lúc nào cũng là lựa chọn phù hợp nhất. Ví dụ, các transporter kiểu streaming như Kafka hoặc NATS streaming, dùng cơ chế lưu trữ dựa trên log, được tối ưu để giải quyết một nhóm vấn đề khác, gần với mô hình event messaging hơn (xem event-based messaging để biết chi tiết).

Để bật kiểu message request-response, Nest tạo ra hai channel logic: một để truyền data, một để chờ response trả về. Với một số transport bên dưới, như NATS, việc hỗ trợ hai channel này đã có sẵn. Với các transport khác, Nest bù đắp bằng cách tự tạo thêm channel riêng. Cách này hiệu quả, nhưng có thể phát sinh thêm overhead. Vì vậy, nếu bạn không cần kiểu message request-response, bạn nên xem xét dùng phương thức event-based.

Để tạo một message handler theo mô hình request-response, dùng decorator @MessagePattern(), được import từ package @nestjs/microservices. Decorator này chỉ nên dùng trong các class controller, vì controller đóng vai trò là entry point của ứng dụng. Dùng nó trong provider sẽ không có tác dụng, vì Nest runtime sẽ bỏ qua.

import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';

@Controller()
export class MathController {
  @MessagePattern({ cmd: 'sum' })
  accumulate(data: number[]): number {
    return (data || []).reduce((a, b) => a + b);
  }
}

import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';

@Controller()
export class MathController {
  @MessagePattern({ cmd: 'sum' })
  accumulate(data) {
    return (data || []).reduce((a, b) => a + b);
  }
}

Trong đoạn code trên, message handler accumulate() lắng nghe các message khớp với pattern { cmd: 'sum' }. Message handler nhận một tham số duy nhất là data, được truyền từ client. Trong trường hợp này, data là một array số cần được cộng lại.

Response bất đồng bộ (Asynchronous responses)

Message handler có thể trả response đồng bộ hoặc bất đồng bộ, nghĩa là method async được hỗ trợ.

@MessagePattern({ cmd: 'sum' })
async accumulate(data: number[]): Promise<number> {
  return (data || []).reduce((a, b) => a + b);
}

@MessagePattern({ cmd: 'sum' })
async accumulate(data) {
  return (data || []).reduce((a, b) => a + b);
}

Message handler cũng có thể trả về một Observable, trong trường hợp đó các giá trị kết quả sẽ được emit liên tục cho tới khi stream hoàn tất.

@MessagePattern({ cmd: 'sum' })
accumulate(data: number[]): Observable<number> {
  return from([1, 2, 3]);
}

@MessagePattern({ cmd: 'sum' })
accumulate(data: number[]): Observable<number> {
  return from([1, 2, 3]);
}

Trong ví dụ trên, message handler sẽ trả response ba lần, mỗi lần cho một phần tử trong array.

Event-based

Trong khi phương thức request-response rất phù hợp để trao đổi message giữa các service, nó lại kém phù hợp cho messaging kiểu event-based — khi bạn chỉ đơn giản muốn publish event mà không cần chờ response. Trong trường hợp này, phần overhead của việc duy trì hai channel cho request-response là không cần thiết.

Ví dụ, nếu bạn muốn thông báo cho một service khác biết rằng một điều kiện cụ thể đã xảy ra ở phần này của hệ thống, kiểu message event-based là lựa chọn lý tưởng.

Để tạo một event handler, bạn có thể dùng decorator @EventPattern(), được import từ package @nestjs/microservices.

@EventPattern('user_created')
async handleUserCreated(data: Record<string, unknown>) {
  // business logic
}

@EventPattern('user_created')
async handleUserCreated(data) {
  // business logic
}

Event handler handleUserCreated() lắng nghe event 'user_created'. Event handler nhận một tham số duy nhất là data, được truyền từ client (trong trường hợp này là payload của event, đã được gửi qua network).

Thông tin bổ sung về request (Additional request details)

Trong các tình huống phức tạp hơn, bạn có thể cần truy cập thêm thông tin về request đến. Ví dụ, khi dùng NATS với wildcard subscription, bạn có thể muốn lấy subject gốc mà producer đã gửi message tới. Tương tự, với Kafka, bạn có thể cần truy cập message headers. Để làm điều này, bạn có thể dùng các decorator có sẵn như dưới đây:

@MessagePattern('time.us.*')
getDate(@Payload() data: number[], @Ctx() context: NatsContext) {
  console.log(`Subject: ${context.getSubject()}`); // e.g. "time.us.east"
  return new Date().toLocaleTimeString(...);
}

@Bind(Payload(), Ctx())
@MessagePattern('time.us.*')
getDate(data, context) {
  console.log(`Subject: ${context.getSubject()}`); // e.g. "time.us.east"
  return new Date().toLocaleTimeString(...);
}

Client (producer class)

Một ứng dụng Nest đóng vai trò client có thể trao đổi message hoặc publish event tới một Nest microservice bằng class ClientProxy. Class này cung cấp nhiều method, như send() (cho messaging kiểu request-response) và emit() (cho messaging kiểu event-driven), giúp giao tiếp với một microservice từ xa. Bạn có thể lấy một instance của class này theo các cách sau:

Một cách là import ClientsModule, module này expose static method register(). Method này nhận một array các object đại diện cho transporter của microservice. Mỗi object phải có property name, và có thể có thêm property transport (mặc định là Transport.TCP), cũng như property options (tùy chọn).

Property name đóng vai trò là một injection token, bạn có thể dùng nó để inject một instance ClientProxy ở bất cứ đâu cần. Giá trị của property name này có thể là một string hoặc JavaScript symbol tùy ý, như mô tả tại đây.

Property options là một object chứa các thuộc tính tương tự như những gì ta thấy trong method createMicroservice() ở trên.

@Module({
  imports: [
    ClientsModule.register([
      { name: 'MATH_SERVICE', transport: Transport.TCP },
    ]),
  ],
})

Ngoài ra, bạn có thể dùng method registerAsync() nếu cần cung cấp configuration hoặc thực hiện các xử lý bất đồng bộ khác trong quá trình setup.

@Module({
  imports: [
    ClientsModule.registerAsync([
      {
        imports: [ConfigModule],
        name: 'MATH_SERVICE',
        useFactory: async (configService: ConfigService) => ({
          transport: Transport.TCP,
          options: {
            url: configService.get('URL'),
          },
        }),
        inject: [ConfigService],
      },
    ]),
  ],
})

Sau khi import module, bạn có thể inject một instance ClientProxy đã được cấu hình với các option của transporter 'MATH_SERVICE' bằng decorator @Inject().

constructor(
  @Inject('MATH_SERVICE') private client: ClientProxy,
) {}

Đôi khi, bạn có thể cần lấy configuration của transporter từ một service khác (chẳng hạn ConfigService), thay vì hard-code trong ứng dụng client. Để làm điều này, bạn có thể đăng ký một custom provider dùng class ClientProxyFactory. Class này cung cấp static method create() nhận vào một object option của transporter và trả về một instance ClientProxy đã được tùy chỉnh.

@Module({
  providers: [
    {
      provide: 'MATH_SERVICE',
      useFactory: (configService: ConfigService) => {
        const mathSvcOptions = configService.getMathSvcOptions();
        return ClientProxyFactory.create(mathSvcOptions);
      },
      inject: [ConfigService],
    }
  ]
  ...
})

Một lựa chọn khác là dùng property decorator @Client().

@Client({ transport: Transport.TCP })
client: ClientProxy;

Dùng decorator @Client() không phải là kỹ thuật được khuyến nghị, vì nó khó test hơn và khó chia sẻ chung một client instance hơn.

ClientProxy là lazy. Nó không khởi tạo kết nối ngay lập tức. Thay vào đó, kết nối sẽ được thiết lập trước lần gọi microservice đầu tiên, sau đó được tái sử dụng cho mỗi lần gọi tiếp theo. Tuy nhiên, nếu bạn muốn trì hoãn quá trình bootstrap ứng dụng cho đến khi kết nối được thiết lập, bạn có thể tự chủ động gọi method connect() của object ClientProxy trong lifecycle hook OnApplicationBootstrap.

async onApplicationBootstrap() {
  await this.client.connect();
}

Nếu không thể tạo được kết nối, method connect() sẽ reject với error object tương ứng.

Gửi message (Sending messages)

ClientProxy expose method send(). Method này dùng để gọi microservice và trả về một Observable chứa response. Nhờ vậy, ta có thể subscribe vào các giá trị được emit một cách dễ dàng.

accumulate(): Observable<number> {
  const pattern = { cmd: 'sum' };
  const payload = [1, 2, 3];
  return this.client.send<number>(pattern, payload);
}

accumulate() {
  const pattern = { cmd: 'sum' };
  const payload = [1, 2, 3];
  return this.client.send(pattern, payload);
}

Method send() nhận hai tham số, pattern và payload. pattern phải khớp với pattern đã định nghĩa trong decorator @MessagePattern(). payload là message mà ta muốn truyền tới microservice từ xa. Method này trả về một Observable lạnh (cold), nghĩa là bạn phải chủ động subscribe vào nó trước khi message được gửi đi.

Publish event (Publishing events)

Để gửi một event, dùng method emit() của object ClientProxy. Method này publish một event tới message broker.

async publish() {
  this.client.emit<number>('user_created', new UserCreatedEvent());
}

async publish() {
  this.client.emit('user_created', new UserCreatedEvent());
}

Method emit() nhận hai tham số: pattern và payload. pattern phải khớp với pattern đã định nghĩa trong decorator @EventPattern(), còn payload là dữ liệu của event mà bạn muốn truyền tới microservice từ xa. Method này trả về một Observable nóng (hot) (khác với Observable lạnh mà send() trả về), nghĩa là bất kể bạn có subscribe vào observable hay không, proxy vẫn sẽ ngay lập tức thử gửi event đi.

Phạm vi theo request (Request-scoping)

Với những ai quen làm việc với các ngôn ngữ lập trình khác, có thể sẽ bất ngờ khi biết rằng trong Nest, phần lớn mọi thứ được chia sẻ (shared) giữa các request đến. Điều này bao gồm connection pool tới database, các service singleton có global state, và nhiều thứ khác. Hãy nhớ rằng Node.js không theo mô hình request/response đa luồng (multi-threaded) và stateless, nơi mỗi request được xử lý bởi một thread riêng. Vì vậy, dùng singleton instance là an toàn cho ứng dụng của chúng ta.

Tuy nhiên, có những trường hợp đặc biệt (edge case) mà việc handler có lifetime theo từng request lại là điều mong muốn. Ví dụ như caching theo từng request trong ứng dụng GraphQL, request tracking, hoặc multi-tenancy. Bạn có thể tìm hiểu thêm về cách kiểm soát scope tại đây.

Handler và provider có scope request có thể inject RequestContext bằng decorator @Inject() kết hợp với token CONTEXT:

import { Injectable, Scope, Inject } from '@nestjs/common';
import { CONTEXT, RequestContext } from '@nestjs/microservices';

@Injectable({ scope: Scope.REQUEST })
export class CatsService {
  constructor(@Inject(CONTEXT) private ctx: RequestContext) {}
}

Điều này cho phép truy cập object RequestContext, có hai property:

export interface RequestContext<T = any> {
  pattern: string | Record<string, any>;
  data: T;
}

Property data là message payload được gửi bởi bên sản xuất message (producer). Property pattern là pattern được dùng để xác định handler phù hợp để xử lý message đến.

Cập nhật trạng thái instance (Instance status updates)

Để nhận cập nhật theo thời gian thực về kết nối và trạng thái của driver instance bên dưới, bạn có thể subscribe vào stream status. Stream này cung cấp các cập nhật trạng thái riêng cho từng driver được chọn. Ví dụ, nếu bạn dùng transporter TCP (mặc định), stream status sẽ emit event connected và disconnected.

this.client.status.subscribe((status: TcpStatus) => {
  console.log(status);
});

Tương tự, bạn có thể subscribe vào stream status của server để nhận thông báo về trạng thái của server.

const server = app.connectMicroservice<MicroserviceOptions>(...);
server.status.subscribe((status: TcpStatus) => {
  console.log(status);
});

Lắng nghe các event nội bộ (Listening to internal events)

Trong một số trường hợp, bạn có thể muốn lắng nghe các event nội bộ được emit bởi microservice. Ví dụ, bạn có thể lắng nghe event error để kích hoạt thêm các xử lý khác khi có lỗi xảy ra. Để làm điều này, dùng method on(), như dưới đây:

this.client.on('error', (err) => {
  console.error(err);
});

Tương tự, bạn có thể lắng nghe các event nội bộ của server:

server.on<TcpEvents>('error', (err) => {
  console.error(err);
});

Truy cập driver bên dưới (Underlying driver access)

Với các use case phức tạp hơn, bạn có thể cần truy cập driver instance bên dưới. Điều này hữu ích cho các tình huống như chủ động đóng kết nối hoặc dùng các method riêng của driver. Tuy nhiên, hãy nhớ rằng trong hầu hết trường hợp, bạn không nên cần truy cập trực tiếp driver.

Để làm điều này, bạn có thể dùng method unwrap(), trả về driver instance bên dưới. Generic type parameter nên chỉ định type của driver instance mà bạn mong đợi.

const netServer = this.client.unwrap<Server>();

Ở đây, Server là một type được import từ module net.

Tương tự, bạn có thể truy cập driver instance bên dưới của server:

const netServer = server.unwrap<Server>();

Xử lý timeout (Handling timeouts)

Trong hệ thống phân tán, microservices đôi khi có thể bị down hoặc không khả dụng. Để tránh chờ đợi vô thời hạn, bạn có thể dùng timeout. Timeout là một pattern rất hữu ích khi giao tiếp với các service khác. Để áp dụng timeout cho các lệnh gọi microservice, bạn có thể dùng operator timeout của RxJS. Nếu microservice không trả response trong khoảng thời gian quy định, một exception sẽ được throw, và bạn có thể catch để xử lý phù hợp.

Để triển khai điều này, bạn cần dùng package rxjs. Chỉ cần dùng operator timeout trong pipe:

this.client
  .send<TResult, TInput>(pattern, data)
  .pipe(timeout(5000));

this.client
  .send(pattern, data)
  .pipe(timeout(5000));

Sau 5 giây, nếu microservice không phản hồi, nó sẽ throw một lỗi.

Trace một request đi qua nhiều service (Tracing a request across services)

Timeout cho bạn biết một lệnh gọi không kịp trả về đúng hạn. Nhưng nó không cho biết thời gian đã trôi vào đâu — và trong một hệ thống gồm năm service giao tiếp qua TCP, NATS, và Kafka, đó mới chính là câu hỏi đáng đặt ra. Một request tới gateway mất 3 giây có thể tiêu tốn 2.9 giây trong đó ở một service downstream mà không ai ngờ tới, và log riêng của từng service, khi xem một mình, đều sẽ khẳng định là mọi thứ vẫn ổn.

Cách khắc phục thông thường là tự truyền tay một correlation id qua từng transport, rồi ghép lại timeline sau đó. NestJS Observe làm việc ghép nối này cho bạn: instrument từng service với SDK @nestjs/observe và forward trace id qua bất kỳ channel nào transport đang sẵn có — một header Kafka, một header NATS, một field trong payload TCP — rồi dashboard sẽ tái tạo lại một waterfall duy nhất trải dài qua mọi service đã tham gia:

Từ đó, timeout không còn là một bí ẩn nữa. Bạn có thể thấy send() ở gateway đang chờ, consumer nhận message (và nó đã “nằm chờ” bao lâu trước khi được xử lý), câu query bên trong nó chạy lâu, và lỗi mà nó cuối cùng throw ra, kèm dòng code nguồn — tất cả trên cùng một mốc thời gian. Message handler và event handler được instrument tự động, nên các handler @MessagePattern() và @EventPattern() hiện ra như các operation mà không cần khai báo span thủ công.

Việc forward trace id là phần duy nhất cần code trong ứng dụng, vì chỉ bạn mới biết transport của mình còn “chỗ trống” nào để tận dụng. Xem Distributed tracing để biết pattern cho từng transport, và chương Observability để thiết lập.

Hỗ trợ TLS (TLS support)

Khi giao tiếp bên ngoài một private network, việc mã hóa traffic để đảm bảo an toàn là rất quan trọng. Trong NestJS, điều này có thể thực hiện bằng TLS qua TCP, dùng module TLS có sẵn của Node. Nest hỗ trợ sẵn TLS trong transport TCP, cho phép ta mã hóa giao tiếp giữa các microservice hoặc client.

Để bật TLS cho một TCP server, bạn cần một private key và một certificate ở dạng PEM. Các file này được thêm vào option của server bằng cách thiết lập tlsOptions và chỉ định file key, cert, như dưới đây:

import * as fs from 'node:fs';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module.js';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';

async function bootstrap() {
  const key = fs.readFileSync('<pathToKeyFile>', 'utf8').toString();
  const cert = fs.readFileSync('<pathToCertFile>', 'utf8').toString();

  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.TCP,
      options: {
        tlsOptions: {
          key,
          cert,
        },
      },
    },
  );

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

Để một client có thể giao tiếp an toàn qua TLS, ta cũng định nghĩa object tlsOptions, nhưng lần này với CA certificate. Đây là certificate của tổ chức đã ký certificate của server. Điều này đảm bảo client tin tưởng certificate của server và có thể thiết lập một kết nối an toàn.

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MATH_SERVICE',
        transport: Transport.TCP,
        options: {
          tlsOptions: {
            ca: [fs.readFileSync('<pathToCaFile>', 'utf-8').toString()],
          },
        },
      },
    ]),
  ],
})
export class AppModule {}

Bạn cũng có thể truyền một array các CA nếu hệ thống của bạn liên quan tới nhiều tổ chức tin cậy khác nhau.

Sau khi mọi thứ đã được thiết lập, bạn có thể inject ClientProxy như bình thường bằng decorator @Inject() để dùng client trong service của mình. Điều này đảm bảo giao tiếp được mã hóa xuyên suốt các microservice NestJS của bạn, với module TLS của Node xử lý phần mã hóa.

Để biết thêm thông tin, xem tài liệu TLS của Node.

Cấu hình động (Dynamic configuration)

Khi một microservice cần được cấu hình bằng ConfigService (từ package @nestjs/config), nhưng injection context chỉ có sẵn sau khi microservice instance đã được tạo, AsyncMicroserviceOptions cung cấp một giải pháp. Cách tiếp cận này cho phép cấu hình động, đảm bảo tích hợp trơn tru với ConfigService.

import { ConfigService } from '@nestjs/config';
import { AsyncMicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module.js';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<AsyncMicroserviceOptions>(
    AppModule,
    {
      useFactory: (configService: ConfigService) => ({
        transport: Transport.TCP,
        options: {
          host: configService.get<string>('HOST'),
          port: configService.get<number>('PORT'),
        },
      }),
      inject: [ConfigService],
    },
  );

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