RabbitMQ

RabbitMQ - bir nechta messaging protokollarini qo'llab-quvvatlaydigan open source va yengil message broker. U yuqori masshtab va yuqori mavjudlik talablarini qondirish uchun taqsimlangan va federatsiyalangan konfiguratsiyalarda joylashtirilishi mumkin. Bundan tashqari, u dunyo bo'ylab kichik startaplardan katta korxonalargacha eng keng tarqalgan message broker hisoblanadi.

O'rnatish

RabbitMQ asosidagi microservice'larni qurishni boshlash uchun avval kerakli paketlarni o'rnating:

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

Umumiy ko'rinish

RabbitMQ transportyoridan foydalanish uchun createMicroservice() metodiga quyidagi options obyektini uzating:

1const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
2  transport: Transport.RMQ,
3  options: {
4    urls: ['amqp://localhost:5672'],
5    queue: 'cats_queue',
6    queueOptions: {
7      durable: false
8    },
9  },
10});

Opsiyalar

options propertysi tanlangan transportyorga xos. RabbitMQ transportyori quyida tasvirlangan propertylarni ochadi.

Mijoz

Boshqa microservice transportyorlari kabi, RabbitMQ ClientProxy instansiyasini yaratish uchun bir nechta variant bor.

Instansiya yaratishning bir usuli - ClientsModuledan foydalanish. ClientsModule orqali client instansiyasini yaratish uchun uni import qiling va register() metodiga createMicroservice() metodida yuqorida ko'rsatilgan xuddi shu propertylarga ega options obyektini, shuningdek injection token sifatida ishlatiladigan name propertysini uzating. ClientsModule haqida batafsil hereda o'qing.

1@Module({
2  imports: [
3    ClientsModule.register([
4      {
5        name: 'MATH_SERVICE',
6        transport: Transport.RMQ,
7        options: {
8          urls: ['amqp://localhost:5672'],
9          queue: 'cats_queue',
10          queueOptions: {
11            durable: false
12          },
13        },
14      },
15    ]),
16  ]
17  ...
18})

Client yaratishning boshqa usullari (ClientProxyFactory yoki @Client()) ham qo'llanishi mumkin. Ular haqida hereda o'qishingiz mumkin.

Kontekst

Murakkabroq ssenariylarda kiruvchi so'rov haqida qo'shimcha ma'lumotlarga kirish kerak bo'lishi mumkin. RabbitMQ transportyoridan foydalanganda RmqContext obyektiga kirishingiz mumkin.

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

RabbitMQ xabarining asl nusxasiga (properties, fields va content bilan) kirish uchun RmqContext obyektining getMessage() metodidan quyidagicha foydalaning:

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

RabbitMQ channeliga referens olish uchun RmqContext obyektining getChannelRef metodidan quyidagicha foydalaning:

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

Xabarni tasdiqlash (acknowledgement)

Xabar hech qachon yo'qolmasligini ta'minlash uchun RabbitMQ message acknowledgementsni qo'llab-quvvatlaydi. Acknowledgement consumer tomonidan RabbitMQga yuborilib, xabar olingani, qayta ishlangani va RabbitMQ xabarni o'chirishi mumkinligi haqida xabar beradi. Agar consumer o'lsa (kanal yopilsa, ulanish yopilsa yoki TCP ulanish uzilsa) va ack yuborilmasa, RabbitMQ xabar to'liq qayta ishlanmaganini tushunadi va uni qayta queuega qo'yadi.

Qo'lda acknowledgement rejimini yoqish uchun noAck propertysini false ga o'rnating:

1options: {
2  urls: ['amqp://localhost:5672'],
3  queue: 'cats_queue',
4  noAck: false,
5  queueOptions: {
6    durable: false
7  },
8},

Qo'lda consumer acknowledgementlari yoqilganda, ishchi tomonidan to'g'ri acknowledgement yuborib, vazifa tugaganini bildirishimiz kerak.

1@MessagePattern('notifications')
2getNotifications(@Payload() data: number[], @Ctx() context: RmqContext) {
3  const channel = context.getChannelRef();
4  const originalMsg = context.getMessage();
5
6  channel.ack(originalMsg);
7}

Record builderlari

Xabar options'larini sozlash uchun RmqRecordBuilder klassidan foydalanishingiz mumkin (eslatma: bu event-based flowlar uchun ham mumkin). Masalan, headers va priority propertylarini o'rnatish uchun setOptions metodidan quyidagicha foydalaning:

1const message = ':cat:';
2const record = new RmqRecordBuilder(message)
3  .setOptions({
4    headers: {
5      ['x-version']: '1.0.0',
6    },
7    priority: 3,
8  })
9  .build();
10
11this.client.send('replace-emoji', record).subscribe(...);

Bu qiymatlarni server tomonida ham RmqContextga kirish orqali quyidagicha o'qishingiz mumkin:

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

Instansiya holati yangilanishlari

Ulanish va asosiy driver instansiyasi holati haqida real vaqt yangilanishlarini olish uchun status streamiga subscribe bo'lishingiz mumkin. Bu stream tanlangan driverga xos holat yangilanishlarini beradi. RMQ driveri uchun status streami connected va disconnected eventlarini emit qiladi.

1this.client.status.subscribe((status: RmqStatus) => {
2  console.log(status);
3});

Xuddi shuningdek, serverning status streamiga subscribe bo'lib, server holati haqida xabarlarni olishingiz mumkin.

1const server = app.connectMicroservice<MicroserviceOptions>(...);
2server.status.subscribe((status: RmqStatus) => {
3  console.log(status);
4});

RabbitMQ eventlarini tinglash

Ba'zi holatlarda microservice tomonidan emit qilinadigan ichki eventlarni tinglashni xohlashingiz mumkin. Masalan, xato yuz berganda qo'shimcha operatsiyalarni ishga tushirish uchun error eventini tinglashingiz mumkin. Buning uchun quyida ko'rsatilgandek on() metodidan foydalaning:

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

Xuddi shuningdek, serverning ichki eventlarini tinglashingiz mumkin:

1server.on<RmqEvents>('error', (err) => {
2  console.error(err);
3});

Asosiy driverga kirish

Murakkabroq use-case'larda asosiy driver instansiyasiga kirish kerak bo'lishi mumkin. Bu ulanishni qo'lda yopish yoki driverga xos metodlardan foydalanish kabi ssenariylar uchun foydali. Biroq, ko'p hollarda driverga to'g'ridan-to'g'ri kirish kerak emasligini yodda tuting.

Buning uchun unwrap() metodidan foydalanishingiz mumkin, u asosiy driver instansiyasini qaytaradi. Generic type parametri kutilayotgan driver instansiyasi turini ko'rsatishi kerak.

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

Xuddi shuningdek, serverning asosiy driver instansiyasiga kirishingiz mumkin:

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

Wildcardlar

RabbitMQ routing keylarda wildcardlardan foydalanishni qo'llab-quvvatlaydi, bu xabarlarni moslashuvchan routlashga imkon beradi. # wildcardi nol yoki undan ko'p so'zlarni, * wildcardi esa aynan bitta so'zni moslaydi.

Masalan, cats.# routing keyi cats, cats.meow va cats.meow.purrni moslaydi. cats.* routing keyi cats.meowni moslaydi, ammo cats.meow.purrni moslamaydi.

RabbitMQ microservice'ingizda wildcardlarni yoqish uchun options obyektida wildcards konfiguratsiya optionini true ga o'rnating:

1const app = await NestFactory.createMicroservice<MicroserviceOptions>(
2  AppModule,
3  {
4    transport: Transport.RMQ,
5    options: {
6      urls: ['amqp://localhost:5672'],
7      queue: 'cats_queue',
8      wildcards: true,
9    },
10  },
11);

Ushbu konfiguratsiya bilan event/message'ga subscribe bo'lganda routing keylarda wildcardlardan foydalanishingiz mumkin. Masalan, cats.# routing keyiga ega xabarlarni tinglash uchun quyidagi koddan foydalaning:

1@MessagePattern('cats.#')
2getCats(@Payload() data: { message: string }, @Ctx() context: RmqContext) {
3  console.log(`Received message with routing key: ${context.getPattern()}`);
4
5  return {
6    message: 'Hello from the cats service!',
7  }
8}

Ma'lum routing key bilan xabar yuborish uchun ClientProxy instansiyasining send() metodidan foydalanishingiz mumkin:

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