الفصل 3: المراسلة غير المتزامنة مع RabbitMQ

أهداف الفصل

• فهم مفاهيم المراسلة غير المتزامنة
• إعداد RabbitMQ مع Spring Boot
• تنفيذ المنتجين والمستهلكين
• إدارة التبادلات، الطوابير والربط
• تنفيذ أنماط المراسلة

---

1. مقدمة في المراسلة

1.1 لماذا المراسلة غير المتزامنة

متزامن (REST):
الخدمة A --HTTP--> الخدمة B --HTTP--> الخدمة C
    └── انتظار الرد ──┘     └── انتظار الرد ──┘
    
مشاكل:
- اقتران زمني
- تعطل الخدمة B = فشل الخدمة A
- تراكم التأخير

غير متزامن (المراسلة):
الخدمة A --نشر--> [طابور] --استهلاك--> الخدمة B
    └── متابعة فورية

المزايا:
- فك الاقتران
- المرونة
- القابلية للتوسع
- معالجة الذروة

1.2 مفاهيم RabbitMQ

المفهوم	الوصف
المنتج	يرسل الرسائل
المستهلك	يستقبل الرسائل
الطابور	قائمة انتظار الرسائل
التبادل	يوجه الرسائل إلى الطوابير
الربط	قاعدة توجيه التبادل -> الطابور
مفتاح التوجيه	مفتاح للتوجيه

1.3 أنواع التبادل

التبادل المباشر:
المنتج --> [التبادل] --> الطابور (تطابق مفتاح التوجيه الدقيق)

التبادل الموزع:
المنتج --> [التبادل] --> الطابور1
                      └--> الطابور2
                      └--> الطابور3

التبادل الموضوعي:
المنتج --> [التبادل] --> الطابور (النمط: order.*, *.created)

التبادل الرأسي:
المنتج --> [التبادل] --> الطابور (مطابقة الرأس)

---

2. إعداد Spring AMQP

2.1 تبعيات Maven

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

2.2 إعداد application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    
    # إعداد المستمع
    listener:
      simple:
        acknowledge-mode: auto
        concurrency: 3
        max-concurrency: 10
        prefetch: 10
        retry:
          enabled: true
          initial-interval: 1000
          max-attempts: 3
          max-interval: 10000
          multiplier: 2
    
    # إعداد القالب
    template:
      retry:
        enabled: true
        initial-interval: 1000
        max-attempts: 3

2.3 إعداد Java

package com.example.messaging.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    
    // أسماء الطوابير والتبادلات
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_ROUTING_KEY = "order.created";
    
    public static final String NOTIFICATION_QUEUE = "notification.queue";
    public static final String NOTIFICATION_EXCHANGE = "notification.exchange";
    
    // طابور للطلبات
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "dlx.order")
            .build();
    }
    
    // تبادل مباشر للطلبات
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    
    // ربط طابور الطلبات <-> التبادل
    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder
            .bind(orderQueue)
            .to(orderExchange)
            .with(ORDER_ROUTING_KEY);
    }
    
    // طابور للإشعارات
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }
    
    // تبادل موزع للإشعارات (بث)
    @Bean
    public FanoutExchange notificationExchange() {
        return new FanoutExchange(NOTIFICATION_EXCHANGE);
    }
    
    // ربط الإشعارات
    @Bean
    public Binding notificationBinding(Queue notificationQueue, 
            FanoutExchange notificationExchange) {
        return BindingBuilder
            .bind(notificationQueue)
            .to(notificationExchange);
    }
    
    // محول JSON
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    // قالب مُعد
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }
}

---

3. المنتج (إرسال الرسائل)

3.1 رسالة DTO

package com.example.messaging.dto;

import java.io.Serializable;
import java.time.LocalDateTime;

public class OrderEvent implements Serializable {
    
    private Long orderId;
    private Long userId;
    private String productName;
    private Double totalAmount;
    private String status;
    private LocalDateTime createdAt;
    
    public OrderEvent() {
        this.createdAt = LocalDateTime.now();
    }
    
    public OrderEvent(Long orderId, Long userId, String productName, 
            Double totalAmount, String status) {
        this();
        this.orderId = orderId;
        this.userId = userId;
        this.productName = productName;
        this.totalAmount = totalAmount;
        this.status = status;
    }
    
    // Getters و Setters
    public Long getOrderId() { return orderId; }
    public void setOrderId(Long orderId) { this.orderId = orderId; }
    
    public Long getUserId() { return userId; }
    public void setUserId(Long userId) { this.userId = userId; }
    
    public String getProductName() { return productName; }
    public void setProductName(String productName) { this.productName = productName; }
    
    public Double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(Double totalAmount) { this.totalAmount = totalAmount; }
    
    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    @Override
    public String toString() {
        return "OrderEvent{orderId=" + orderId + ", userId=" + userId + 
               ", status=" + status + "}";
    }
}

3.2 خدمة المنتج

package com.example.messaging.service;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderEventProducer {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventProducer.class);
    
    private final RabbitTemplate rabbitTemplate;
    
    public OrderEventProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void sendOrderCreatedEvent(OrderEvent event) {
        log.info("Sending order created event: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Order event sent successfully");
    }
    
    public void sendNotification(String message) {
        log.info("Broadcasting notification: {}", message);
        
        // Fanout - لا حاجة لمفتاح التوجيه
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.NOTIFICATION_EXCHANGE,
            "",
            message
        );
    }
    
    // إرسال مع تأكيد
    public void sendWithConfirmation(OrderEvent event) {
        rabbitTemplate.invoke(operations -> {
            operations.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                event
            );
            
            // انتظار التأكيد
            boolean confirmed = operations.waitForConfirms(5000);
            if (!confirmed) {
                throw new RuntimeException("Message not confirmed by broker");
            }
            
            return null;
        });
    }
}

3.3 الاستخدام في خدمة

@Service
public class OrderService {
    
    private final OrderRepository orderRepository;
    private final OrderEventProducer eventProducer;
    
    public OrderService(OrderRepository orderRepository, 
            OrderEventProducer eventProducer) {
        this.orderRepository = orderRepository;
        this.eventProducer = eventProducer;
    }
    
    @Transactional
    public Order createOrder(OrderRequest request) {
        // 1. إنشاء الطلب
        Order order = new Order();
        order.setUserId(request.getUserId());
        order.setProductName(request.getProductName());
        order.setTotalAmount(request.getTotalAmount());
        order.setStatus("CREATED");
        
        Order saved = orderRepository.save(order);
        
        // 2. نشر الحدث
        OrderEvent event = new OrderEvent(
            saved.getId(),
            saved.getUserId(),
            saved.getProductName(),
            saved.getTotalAmount(),
            saved.getStatus()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return saved;
    }
}

---

4. المستهلك (استقبال الرسائل)

4.1 مستمع بسيط

package com.example.messaging.listener;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventListener.class);
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderEvent(OrderEvent event) {
        log.info("Received order event: {}", event);
        
        try {
            // معالجة الرسالة
            processOrder(event);
            log.info("Order event processed successfully");
        } catch (Exception e) {
            log.error("Error processing order event: {}", e.getMessage());
            throw e; // إعادة الرمي لإعادة المحاولة/DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // منطق العمل
        // - إرسال بريد تأكيد
        // - تحديث المخزون
        // - إخطار نظام التوصيل
        log.info("Processing order {} for user {}", 
            event.getOrderId(), event.getUserId());
    }
}

4.2 مستمع مع تأكيد يدوي

@Component
public class OrderEventListener {
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, 
                    ackMode = "MANUAL")
    public void handleOrderEvent(OrderEvent event, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        
        log.info("Received order event: {}", event);
        
        try {
            processOrder(event);
            
            // تأكيد النجاح
            channel.basicAck(deliveryTag, false);
            log.info("Message acknowledged");
            
        } catch (Exception e) {
            log.error("Error processing: {}", e.getMessage());
            
            try {
                // رفض وإعادة الطابور (أو إرسال إلى DLQ)
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Error rejecting message", ioEx);
            }
        }
    }
}

4.3 مستمع للإشعارات (موزع)

@Component
public class NotificationListener {
    
    private static final Logger log = LoggerFactory.getLogger(NotificationListener.class);
    
    @RabbitListener(queues = RabbitMQConfig.NOTIFICATION_QUEUE)
    public void handleNotification(String message) {
        log.info("Received notification: {}", message);
        // معالجة الإشعار
    }
}

4.4 مستمعون متعددون (التوسع)

@Component
public class OrderEventListener {
    
    // 5 مستهلكين بالتوازي
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE, concurrency = "5")
    public void handleOrderEvent(OrderEvent event) {
        log.info("Processing on thread: {}", Thread.currentThread().getName());
        processOrder(event);
    }
}

---

5. طابور الرسائل الميتة (DLQ)

5.1 إعداد DLQ

@Configuration
public class DeadLetterConfig {
    
    public static final String DLX_EXCHANGE = "dlx.exchange";
    public static final String DLQ_ORDER = "dlq.order";
    
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }
    
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_ORDER).build();
    }
    
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder
            .bind(deadLetterQueue())
            .to(deadLetterExchange())
            .with("dlx.order");
    }
}

5.2 مستمع DLQ

@Component
public class DeadLetterListener {
    
    private static final Logger log = LoggerFactory.getLogger(DeadLetterListener.class);
    
    private final OrderRepository orderRepository;
    private final NotificationService notificationService;
    
    @RabbitListener(queues = "dlq.order")
    public void handleDeadLetter(OrderEvent event, 
            @Header("x-death") List<Map<String, Object>> xDeath) {
        
        log.error("Dead letter received: {}", event);
        
        if (xDeath != null && !xDeath.isEmpty()) {
            Map<String, Object> death = xDeath.get(0);
            log.error("Death reason: {}, count: {}", 
                death.get("reason"), death.get("count"));
        }
        
        // إجراءات ممكنة:
        // 1. حفظ للتحليل
        // 2. إخطار فريق العمليات
        // 3. إعادة المحاولة يدويًا
        
        notificationService.alertOpsTeam(
            "Order processing failed: " + event.getOrderId()
        );
    }
}

---

6. تبادل الموضوع (التوجيه المتقدم)

6.1 الإعداد

@Configuration
public class TopicExchangeConfig {
    
    public static final String TOPIC_EXCHANGE = "events.topic";
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    // طابور لجميع أحداث الطلبات
    @Bean
    public Queue allOrdersQueue() {
        return new Queue("queue.orders.all");
    }
    
    // طابور للطلبات التي تم إنشاؤها فقط
    @Bean
    public Queue createdOrdersQueue() {
        return new Queue("queue.orders.created");
    }
    
    // طابور لجميع الأحداث
    @Bean
    public Queue allEventsQueue() {
        return new Queue("queue.events.all");
    }
    
    @Bean
    public Binding allOrdersBinding(Queue allOrdersQueue, TopicExchange topicExchange) {
        return BindingBuilder
            .bind(allOrdersQueue)
            .to(topicExchange)
            .with("order.*");  // order.created, order.shipped, etc.
    }
    
    @Bean
    public Binding createdOrdersBinding(Queue createdOrdersQueue, 
            TopicExchange topicExchange) {
        return BindingBuilder
            .bind(createdOrdersQueue)
            .to(topicExchange)
            .with("order.created");
    }
    
    @Bean
    public Binding allEventsBinding(Queue allEventsQueue, TopicExchange topicExchange) {
        return BindingBuilder
            .bind(allEventsQueue)
            .to(topicExchange)
            .with("#");  // جميع الرسائل
    }
}

6.2 الاستخدام

@Service
public class EventPublisher {
    
    private final RabbitTemplate rabbitTemplate;
    
    public void publishOrderCreated(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "order.created",  // مفتاح التوجيه
            event
        );
    }
    
    public void publishOrderShipped(OrderEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "order.shipped", 
            event
        );
    }
    
    public void publishPaymentReceived(PaymentEvent event) {
        rabbitTemplate.convertAndSend(
            "events.topic", 
            "payment.received", 
            event
        );
    }
}

---

7. نمط الطلب-الرد

7.1 الإعداد

@Configuration
public class RpcConfig {
    
    public static final String RPC_QUEUE = "rpc.queue";
    public static final String RPC_EXCHANGE = "rpc.exchange";
    
    @Bean
    public Queue rpcQueue() {
        return new Queue(RPC_QUEUE);
    }
    
    @Bean
    public DirectExchange rpcExchange() {
        return new DirectExchange(RPC_EXCHANGE);
    }
    
    @Bean
    public Binding rpcBinding() {
        return BindingBuilder
            .bind(rpcQueue())
            .to(rpcExchange())
            .with("rpc");
    }
}

7.2 عميل RPC

@Service
public class RpcClient {
    
    private final RabbitTemplate rabbitTemplate;
    
    public RpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public String callService(String request) {
        Object response = rabbitTemplate.convertSendAndReceive(
            RpcConfig.RPC_EXCHANGE,
            "rpc",
            request
        );
        
        return response != null ? response.toString() : null;
    }
}

7.3 خادم RPC

@Component
public class RpcServer {
    
    @RabbitListener(queues = RpcConfig.RPC_QUEUE)
    public String handleRequest(String request) {
        log.info("Received RPC request: {}", request);
        
        // معالجة
        String result = processRequest(request);
        
        log.info("Sending RPC response: {}", result);
        return result;
    }
    
    private String processRequest(String request) {
        // منطق العمل
        return "Processed: " + request;
    }
}

---

8. ملخص

أنماط المراسلة

النمط	التبادل	الاستخدام
نقطة إلى نقطة	مباشر	مستهلك واحد
نشر/اشتراك	موزع	بث
التوجيه	موضوع	توجيه بالنمط
طلب/رد	مباشر	RPC

أفضل الممارسات

1. استخدام رسائل متكررة
2. تنفيذ طابور الرسائل الميتة
3. إعداد محاولات إعادة مناسبة
4. مراقبة الطوابير (إدارة RabbitMQ)
5. استخدام التسلسل JSON
6. توثيق صيغ الرسائل
7. اختبار مع رسائل غير صالحة

---

الفصل التالي: المهام المجدولة

العودة إلى الوحدة 14