RabbitMQ

RabbitMQ là một message broker mã nguồn mở, nhẹ, hỗ trợ nhiều messaging protocol. Nó có thể được triển khai theo cấu hình distributed và federated để đáp ứng yêu cầu về khả năng mở rộng cao và tính sẵn sàng cao. Ngoài ra, đây là message broker được sử dụng rộng rãi nhất, dùng ở mọi quy mô từ startup nhỏ đến doanh nghiệp lớn trên toàn thế giới.

Installation (cài đặt)

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

$ npm i --save amqplib amqp-connection-manager

Overview (tổng quan)

Để sử dụng RabbitMQ transporter, truyền object options sau vào method createMicroservice():

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

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

Options (các option)

Property options phụ thuộc vào transporter được chọn. Transporter RabbitMQ cung cấp các property được mô tả dưới đây.

Client

Giống như các transporter khác của microservice, bạn có nhiều cách để tạo instance ClientProxy cho RabbitMQ.

Một cách để tạo instance là dùng ClientsModule. Để tạo client instance với ClientsModule, import nó và dùng method register() để truyền một object options với các property giống như đã trình bày ở method createMicroservice() phía trên, cùng với property name dùng làm injection token. Đọc thêm về ClientsModule tại đây.

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MATH_SERVICE',
        transport: Transport.RMQ,
        options: {
          urls: ['amqp://localhost:5672'],
          queue: 'cats_queue',
          queueOptions: {
            durable: false
          },
        },
      },
    ]),
  ]
  ...
})

Bạn cũng có thể dùng các cách khác để tạo client (ClientProxyFactory hoặc @Client()). Đọc thêm về chúng tại đây.

Context (ngữ cảnh)

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. Khi dùng RabbitMQ transporter, bạn có thể truy cập object RmqContext.

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: RmqContext) {
  console.log(`Pattern: ${context.getPattern()}`);
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  console.log(`Pattern: ${context.getPattern()}`);
}

Để truy cập message gốc của RabbitMQ (gồm properties, fields, và content), dùng method getMessage() của object RmqContext, như sau:

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: RmqContext) {
  console.log(context.getMessage());
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  console.log(context.getMessage());
}

Để lấy reference tới channel của RabbitMQ, dùng method getChannelRef của object RmqContext, như sau:

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: RmqContext) {
  console.log(context.getChannelRef());
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  console.log(context.getChannelRef());
}

Message acknowledgement (xác nhận message)

Để đảm bảo message không bao giờ bị mất, RabbitMQ hỗ trợ message acknowledgement. Acknowledgement là tín hiệu consumer gửi lại để báo cho RabbitMQ biết một message cụ thể đã được nhận, xử lý xong, và RabbitMQ có thể xóa message đó. Nếu consumer “chết” (channel bị đóng, connection bị đóng, hoặc mất kết nối TCP) mà chưa gửi ack, RabbitMQ sẽ hiểu là message chưa được xử lý xong và sẽ re-queue lại.

Để bật chế độ acknowledgment thủ công, đặt property noAck thành false:

options: {
  urls: ['amqp://localhost:5672'],
  queue: 'cats_queue',
  noAck: false,
  queueOptions: {
    durable: false
  },
},

Khi manual consumer acknowledgement được bật, chúng ta phải tự gửi acknowledgement từ worker để báo là đã xử lý xong một task.

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: RmqContext) {
  const channel = context.getChannelRef();
  const originalMsg = context.getMessage();

  channel.ack(originalMsg);
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  const channel = context.getChannelRef();
  const originalMsg = context.getMessage();

  channel.ack(originalMsg);
}

Record builders (bộ dựng record)

Để cấu hình các option của message, bạn có thể dùng class RmqRecordBuilder (lưu ý: điều này cũng áp dụng được cho các luồng dựa trên event). Ví dụ, để đặt property headers và priority, dùng method setOptions như sau:

const message = ':cat:';
const record = new RmqRecordBuilder(message)
  .setOptions({
    headers: {
      ['x-version']: '1.0.0',
    },
    priority: 3,
  })
  .build();

this.client.send('replace-emoji', record).subscribe(...);

Và bạn cũng có thể đọc các giá trị này ở phía server, bằng cách truy cập RmqContext, như sau:

@MessagePattern('replace-emoji')
replaceEmoji(@Payload() data: string, @Ctx() context: RmqContext): string {
  const { properties: { headers } } = context.getMessage();
  return headers['x-version'] === '1.0.0' ? '🐱' : '🐈';
}

@Bind(Payload(), Ctx())
@MessagePattern('replace-emoji')
replaceEmoji(data, context) {
  const { properties: { headers } } = context.getMessage();
  return headers['x-version'] === '1.0.0' ? '🐱' : '🐈';
}

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

Để 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 driver đang dùng. Với driver RMQ, stream status phát ra các event connected và disconnected.

this.client.status.subscribe((status: RmqStatus) => {
  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: RmqStatus) => {
  console.log(status);
});

Listening to RabbitMQ events (lắng nghe event của RabbitMQ)

Trong một số trường hợp, bạn có thể muốn lắng nghe các event nội bộ được microservice phát ra. Ví dụ, bạn có thể lắng nghe event error để thực hiện 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ư sau:

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<RmqEvents>('error', (err) => {
  console.error(err);
});

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

Với các use case phức tạp hơn, bạn có thể cần truy cập trực tiếp underlying driver instance. Điều này hữu ích cho các tình huống như tự đóng kết nối bằng tay 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 cần truy cập trực tiếp driver.

Để làm điều này, bạn dùng method unwrap(), trả về underlying driver instance. Generic type parameter nên chỉ rõ loại driver instance bạn mong muốn.

const managerRef =
  this.client.unwrap<import('amqp-connection-manager').AmqpConnectionManager>();

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

const managerRef =
  server.unwrap<import('amqp-connection-manager').AmqpConnectionManager>();

Wildcards (ký tự đại diện)

RabbitMQ hỗ trợ dùng wildcard trong routing key để định tuyến message một cách linh hoạt. Wildcard # khớp với không hoặc nhiều từ, còn wildcard * khớp với đúng một từ.

Ví dụ, routing key cats.# khớp với cats, cats.meow, và cats.meow.purr. Routing key cats.* khớp với cats.meow nhưng không khớp với cats.meow.purr.

Để bật hỗ trợ wildcard trong microservice RabbitMQ của bạn, đặt config option wildcards thành true trong object options:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(
  AppModule,
  {
    transport: Transport.RMQ,
    options: {
      urls: ['amqp://localhost:5672'],
      queue: 'cats_queue',
      wildcards: true,
    },
  },
);

Với cấu hình này, bạn có thể dùng wildcard trong routing key khi subscribe event/message. Ví dụ, để lắng nghe message với routing key cats.#, bạn có thể dùng code sau:

@MessagePattern('cats.#')
getCats(@Payload() data: { message: string }, @Ctx() context: RmqContext) {
  console.log(`Received message with routing key: ${context.getPattern()}`);

  return {
    message: 'Hello from the cats service!',
  }
}

Để gửi message với một routing key cụ thể, bạn có thể dùng method send() của instance ClientProxy:

this.client.send('cats.meow', { message: 'Meow!' }).subscribe((response) => {
  console.log(response);
});