المرونة والمراسلة والرصد

الرسائل غير المتزامنة

18 دقيقة الدرس 5 من 12

الرسائل غير المتزامنة

كل نظام موزَّع يصطدم في نهاية المطاف بالحاجز ذاته: إذا استدعت الخدمة A الخدمةَ B مباشرةً عبر HTTP وكانت B بطيئة أو متوقفة، فإن A ستصبح بطيئة أو متوقفة أيضًا. التبعية هنا تامة. الرسائل غير المتزامنة تكسر هذا الترابط بوضع وسيط رسائل (message broker) بين الخدمتين. تنشر A رسالة وتواصل عملها فورًا، وتستهلك B الرسالة متى كانت مستعدة. لا تعرف إحداهما عنوان الأخرى ولا حالة تشغيلها ولا حِملها الحالي.

يغطي هذا الدرس المفاهيم والتوصيل البرمجي مع Spring Boot والمفاضلات التشغيلية التي تحتاجها لاتخاذ قرارات تصميمية واعية — لا مجرد نسخ كود يعمل.

ما الذي يقدمه وسيط الرسائل

الوسيط هو خادم (RabbitMQ أو Apache Kafka أو Amazon SQS وغيرها) يقبل الرسائل من المنتجين (producers) ويحتفظ بها حتى يسحبها المستهلكون (consumers) أو يستقبلوها. يوفر الوسيط:

الاستدامة — تبقى الرسائل بعد إعادة تشغيل المستهلك (مع تفعيل المثابرة).
الضغط العكسي — إذا كان المستهلك بطيئًا تتراكم الرسائل في الطابور بدلًا من إفشال المنتج.
البث للمتعددين — يمكن تسليم رسالة واحدة منشورة إلى مستهلكين مستقلين متعددين.
الفصل الزمني — لا يحتاج المنتج والمستهلك إلى التشغيل في الوقت ذاته.

النموذج الذهني الأساسي: HTTP كالمكالمة الهاتفية — يجب أن يكون الطرفان متاحَين في آنٍ واحد. أما الرسائل فكالبريد — يُسقط المرسل الرسالة ويقرأها المستقبل لاحقًا. كلا النموذجين صحيح لمشاكل مختلفة.

المفاهيم الأساسية: التبادلات والطوابير والمواضيع

يستخدم RabbitMQ نموذج exchange → binding → queue. ينشر المنتج إلى exchange وليس إلى طابور مباشرةً. يوجّه الـ exchange الرسالة إلى طابور واحد أو أكثر استنادًا إلى مفتاح التوجيه وقواعد الربط. تشترك تطبيقات المستهلك في الطوابير. أنواع الـ exchange الشائعة:

Direct — يوجّه إلى الطوابير التي يطابق مفتاح ربطها مفتاح التوجيه تمامًا.
Topic — مفاتيح التوجيه أنماط مفصولة بنقاط؛ * تطابق كلمة واحدة، # تطابق صفرًا أو أكثر.
Fanout — يتجاهل مفاتيح التوجيه ويبث إلى كل الطوابير المرتبطة.

يستخدم Kafka مفردات مختلفة — topics وpartitions — لكن مبدأ الفصل متطابق. نركز هنا على RabbitMQ عبر Spring AMQP؛ ويخصّص Kafka درسًا مستقلًا.

إضافة Spring AMQP إلى مشروع Spring Boot 3

أضف المبدئ (starter) إلى pom.xml:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

اضبط اتصال الوسيط في application.yml:

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual   # نؤكد يدويًا بعد المعالجة
        prefetch: 5                # الحد الأقصى للرسائل غير المؤكدة لكل مستهلك

استخدم acknowledge-mode: manual في الإنتاج. مع التأكيد التلقائي يحذف الوسيط الرسالة فور تسليمها حتى لو رمى الكود استثناءً. التأكيد اليدوي يتيح لك التأكيد بعد المعالجة الناجحة، فتُعاد الرسائل غير المعالجة إلى الطابور تلقائيًا.

تعريف البنية التحتية كـ Beans

يمكن لـ Spring AMQP تعريف الـ exchange والطابور والـ binding تلقائيًا عند بدء التشغيل. عرّفها كـ beans في فئة إعداد:

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    public static final String ORDER_EXCHANGE   = "orders.exchange";
    public static final String ORDER_QUEUE      = "orders.created";
    public static final String ROUTING_KEY      = "order.created";

    @Bean
    TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE, true, false);
        // durable=true: يبقى بعد إعادة تشغيل الوسيط
        // autoDelete=false: لا يُحذف عند قطع آخر مستهلك
    }

    @Bean
    Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
                .withArgument("x-dead-letter-exchange", "orders.dlx") // تبادل الرسائل الميتة
                .build();
    }

    @Bean
    Binding orderBinding(Queue orderQueue, TopicExchange orderExchange) {
        return BindingBuilder
                .bind(orderQueue)
                .to(orderExchange)
                .with(ROUTING_KEY);
    }
}

تبادل الرسائل الميتة (DLX): عندما تُرفض رسالة أو تنتهي صلاحيتها، يوجّهها الوسيط إلى الـ DLX بدلًا من حذفها صامتًا. اضبط دائمًا DLX لطوابير الإنتاج حتى تكون الرسائل الفاشلة قابلة للفحص لا مفقودة.

إنتاج رسالة

RabbitTemplate هو المكوّن المركزي الآمن للخيوط (thread-safe) للإرسال. أحقنه وانشر:

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    private final RabbitTemplate rabbitTemplate;

    public OrderService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void placeOrder(Order order) {
        // احفظ الطلب في قاعدة البيانات أولًا ...
        rabbitTemplate.convertAndSend(
            RabbitConfig.ORDER_EXCHANGE,
            RabbitConfig.ROUTING_KEY,
            order              // تحوّله Jackson إلى JSON تلقائيًا
        );
        // تعود الدالة فورًا — دون انتظار أي مستهلك
    }
}

بشكل افتراضي تستخدم convertAndSend تسلسل Java. تجاوز ذلك بتعريف Bean من نوع MessageConverter:

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    // ... beans التبادل والطابور والربط أعلاه ...

    @Bean
    Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

استهلاك رسالة

زيّن دالة بـ @RabbitListener ويُنشئ Spring حاوية مستمعة تستطلع الطابور في تجمّع خيوط خلفية:

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import java.io.IOException;

@Service
public class InventoryService {

    @RabbitListener(queues = RabbitConfig.ORDER_QUEUE)
    public void handleOrder(Order order,
                            Channel channel,
                            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
            throws IOException {

        try {
            reserveStock(order);
            channel.basicAck(deliveryTag, false);   // تأكيد: احذف من الطابور
        } catch (InsufficientStockException e) {
            // فشل دائم: أرسل إلى DLX، لا تُعد للطابور
            channel.basicReject(deliveryTag, false);
        } catch (Exception e) {
            // فشل عابر: أعد إلى الطابور للمحاولة مجددًا
            channel.basicNack(deliveryTag, false, true);
        }
    }

    private void reserveStock(Order order) { /* ... */ }
}

الثباتية (Idempotency): أهم عقد المستهلك

لأن الرسائل يمكن تسليمها أكثر من مرة (عطل شبكي، تعطل المستهلك قبل التأكيد)، يجب أن يكون المستهلك ثابتًا (idempotent): معالجة الرسالة ذاتها مرتين يجب أن تُعطي نتيجةً مطابقة لمعالجتها مرة واحدة. استراتيجيات شائعة:

احتفظ بجدول معرّفات الرسائل المعالجة؛ تجاهل ما سبق رؤيته.
استخدم قيودًا فريدة في قاعدة البيانات لتفشل الإدخالات المكررة بصمت.
صمّم العمليات لتكون ثابتةً بطبيعتها (تعيين قيمة لـ X آمن للتكرار؛ زيادة عداد ليست كذلك).

لا تفترض أبدًا التسليم مرة واحدة فقط في النظام الموزع. تضمن مواصفة AMQP التسليم مرة على الأقل. صمّم كل مستهلك ليكون ثابتًا منذ البداية، وإلا ستقضي ساعات في تصحيح رسوم مزدوجة وسجلات مكررة في الإنتاج.

اعتبارات الأمان

وسيط الرسائل هو حد ثقة. احمه:

TLS أثناء النقل — استخدم spring.rabbitmq.ssl.enabled=true مع TLS المتبادل بين الخدمات والوسيط.
بيانات اعتماد لكل خدمة — تحصل كل خدمة مصغّرة على مستخدم RabbitMQ خاص بها بأقل صلاحيات ضرورية (قراءة من طابورها فقط، كتابة إلى تبادلها فقط).
سلامة الرسالة — إذا عبرت الرسائل حدود الثقة، وقّع الحمولة (مثلًا HMAC-SHA256) وتحقق منها قبل المعالجة. وإلا قد يُدخل منتج مارق رسائل احتيالية.
لا بيانات شخصية في مفاتيح التوجيه — يمكن أن تظهر مفاتيح التوجيه في سجلات الوسيط؛ اجعلها هيكلية لا حاملة للبيانات.

الاختيار بين HTTP المتزامن والرسائل غير المتزامنة

لا يسود أيٌّ من النمطين عالميًا. استخدم هذا الحكم التجريبي:

استخدم HTTP حين يحتاج المُستدعي إجابةً فورية (مثلًا استعلام عن سعر منتج قبل عرضه للمستخدم).
استخدم الرسائل حين يحتاج المُستدعي فقط أن يعرف أن العمل قُبِل لا أنه اكتمل (مثلًا تقديم طلب أو إرسال إشعار أو تشغيل تقرير خلفي).

تزيد الرسائل غير المتزامنة المرونةَ والإنتاجيةَ لكنها تضيف تعقيدًا تشغيليًا: تحتاج وسيطًا يعمل ومعالجةً للرسائل الميتة ومراقبةً للمستهلكين ومنطق ثباتية. هذا الثمن يستحق دفعه حين يحتاج نظامك إليه فعلًا.

الخلاصة

تفصل الرسائل غير المتزامنة الخدماتِ في الزمان والمكان: يستدعي المنتج convertAndSend ويواصل عمله؛ يعالج المستهلك بإيقاعه الخاص. تُغلّف Spring AMQP الـ RabbitMQ بـ RabbitTemplate للإرسال و@RabbitListener للاستقبال. استخدم Jackson2JsonMessageConverter للتشغيل البيني، والتأكيدات اليدوية للأمان، وتبادل الرسائل الميتة لرؤية الفشل، وصمّم كل مستهلك ليكون ثابتًا. في الدرس التالي ستتعلم كيف تتحد هذه الأنماط لبناء خدمات مصغّرة مدفوعة بالأحداث بالكامل.