NestJS — Node.js للمؤسسات

وسطاء الرسائل: RabbitMQ و Kafka

16 دقيقة الدرس 38 من 48

وسطاء الرسائل: RabbitMQ و Kafka

حين تحتاج الخدمات إلى التواصل دون اقتران وثيق، يجلس وسيط الرسائل (Message Broker) بينها: المنتجون يرسلون الرسائل إلى الوسيط، والمستهلكون يقرؤون منه بحسب سرعتهم. يشحن NestJS ناقلَيْن أصيلَيْن للوسيطَيْن الأكثر استخدامًا — RabbitMQ (قوائم انتظار AMQP) و Apache Kafka (موضوعات سجل موزّع). يُشكّل اختيار الوسيط الصحيح ضمانات التسليم والإنتاجية والتعقيد التشغيلي في نظامك.

RabbitMQ — المراسلة القائمة على قوائم الانتظار

يُنفّذ RabbitMQ بروتوكول AMQP. ينشر المنتجون الرسائل إلى exchanges؛ يوجّهها الـ exchange إلى قائمة انتظار واحدة أو أكثر؛ يسحب المستهلكون من القائمة ويرسلون إقرارًا (ack) بعد نجاح المعالجة. تُعاد تسليم الرسائل غير المُقرَّة.

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

سجّل ناقل RabbitMQ في main.ts:

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

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.RMQ,
    options: {
      urls: ['amqp://localhost:5672'],
      queue: 'orders_queue',
      queueOptions: { durable: true },   // تبقى بعد إعادة تشغيل الوسيط
      noAck: false,                      // إقرارات يدوية
    },
  });
  await app.listen();
}
bootstrap();

يستخدم معالج المستهلك @MessagePattern (طلب-استجابة) أو @EventPattern (أرسل وانسَ):

import { Controller } from '@nestjs/common';
import { EventPattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';

@Controller()
export class OrdersConsumer {
  @EventPattern('order_placed')
  async handleOrderPlaced(
    @Payload() data: { orderId: string; total: number },
    @Ctx() context: RmqContext,
  ) {
    console.log('Processing order', data.orderId);
    // --- المعالجة ---
    const channel = context.getChannelRef();
    const originalMsg = context.getMessage();
    channel.ack(originalMsg);   // الإقرار فقط بعد النجاح
  }
}

الإقرارات اليدوية تمنع ضياع الرسائل. مع noAck: false، تظل الرسالة في القائمة حتى يُقرّها معالجك صراحةً. إن تعطّلت العملية أثناء التنفيذ، أعاد RabbitMQ تسليمها إلى مستهلك آخر — محقّقًا تسليمًا لمرّة واحدة على الأقل.

Apache Kafka — تدفق الرسائل القائم على الموضوعات

يخزّن Kafka الرسائل في أقسام موضوعات (topic partitions) مرتّبة وغير قابلة للتعديل (سجل إيداع موزّع). ينتمي المستهلكون إلى مجموعة مستهلكين (consumer group)؛ يُخصَّص كل قسم لعضو واحد فقط من المجموعة، مما يُتيح التوسّع الأفقي مع ضمان الترتيب داخل كل قسم. تُحتفظ بالرسائل لمدة قابلة للضبط — ولا تُحذف عند الاستهلاك.

npm install @nestjs/microservices kafkajs

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

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: ['localhost:9092'],
      },
      consumer: {
        groupId: 'orders-consumer-group',   // جميع النسخ تشترك في المجموعة
      },
    },
  });
  await app.listen();
}
bootstrap();

يستخدم الاشتراك في موضوع Kafka مُزخرفَ @EventPattern نفسه:

import { Controller } from '@nestjs/common';
import { EventPattern, Payload, KafkaContext, Ctx } from '@nestjs/microservices';

@Controller()
export class InventoryConsumer {
  @EventPattern('inventory.updated')
  async handleInventoryUpdate(
    @Payload() message: { sku: string; quantity: number },
    @Ctx() context: KafkaContext,
  ) {
    const { offset, partition, topic } = context.getMessage();
    console.log(`[${topic}/${partition}@${offset}]`, message.sku);
    // يُثبَّت الإزاحة تلقائيًا بعد حلّ المعالج (commitOffsets: true افتراضيًا)
  }
}

مجموعات المستهلكين والتوسّع

في Kafka، تشترك جميع نسخ الخدمة في groupId ذاتها. يُخصّص Kafka كل قسم لنسخة واحدة — فإضافة نسخ يرفع الإنتاجية حتى عدد الأقسام. في RabbitMQ، يتنافس المستهلكون المتعددون على القائمة ذاتها بأسلوب الدوران — التوسّع أبسط لكن لا توجد أقسام مرتّبة.

اختر بناءً على ضماناتك. تحتاج قوائم مهام مع تسليم مرّة واحدة أو أكثر، وقوائم رسائل ميتة، ومرونة توجيه؟ استخدم RabbitMQ. تحتاج تدفقات أحداث عالية الإنتاجية، وإعادة تشغيل، وسجلات تدقيق، أو توزيعًا على مجموعات مستهلكين مستقلة متعددة؟ استخدم Kafka. كثير من الأنظمة الإنتاجية تستخدم كليهما.

الاختيار بينهما

RabbitMQ: تكلفة تشغيلية منخفضة، توجيه غني (exchanges مباشرة وموضوعات وبثّ)، دعم ناضج لقوائم الرسائل الميتة، مثالي لقوائع المهام وأنماط الطلب-الاستجابة.
Kafka: إنتاجية عالية جدًا (ملايين رسالة/ثانية)، سجل مرتّب دائم، إعادة تشغيل الرسائل، مجموعات مستهلكين مستقلة متعددة تقرأ من البث ذاته.
كلاهما: يدعمان الإقرارات اليدوية والتسليم لمرّة واحدة على الأقل؛ وكلاهما له دعم أصيل في NestJS بواجهة مُزخرفات متطابقة.

لا تخلط بين قوائم الانتظار والموضوعات. تُسلّم قائمة انتظار RabbitMQ كل رسالة لمستهلك واحد (مستهلكون متنافسون). يمكن أن يستهلك موضوع Kafka مجموعات مستهلكين مستقلة متعددة، تحصل كل منها على نسخة كاملة من كل رسالة. استخدام قائمة انتظار حيث تحتاج إلى توزيع يُسقط الأحداث صامتًا لجميع المستهلكين ما عدا واحد.

الخلاصة

يدعم NestJS كلًّا من RabbitMQ (AMQP، قائمة انتظار، مدفوع بالإقرار) و Kafka (أقسام موضوعات، مجموعات مستهلكين، سجل دائم) عبر مُزخرفات @EventPattern / @MessagePattern متطابقة. يتفوّق RabbitMQ في قوائم المهام والتوجيه؛ ويتألق Kafka في تدفق الأحداث عالي الإنتاجية والإعادة. استخدم دائمًا الإقرارات اليدوية في RabbitMQ لضمان التسليم لمرّة واحدة على الأقل، وجمّع مستهلكي Kafka بصورة صحيحة لتوزيع الأقسام عبر النسخ.