Messaging و Distributed Systems

#Sync و async messaging و دور الـ brokerSynchronous calls, asynchronous messaging and brokers

لو checkout بينتظر الإيميل والتحليلات، تعطل خدمة جانبية ممكن يعطّل شراء سليم. الـ synchronous HTTP مناسب لما محتاج قرار دلوقتي؛ messaging مناسب لشغل يقدر يحصل بعد تثبيت الطلب.

الـ broker بيستقبل ويوجّه ويخزّن حسب إعداداته. مش كل نشر durable، ومش كل قبول معناه إن المستهلك خلّص. فرّق بين command بيطلب فعل من مسؤول معين، و event بيعلن حقيقة حصلت وممكن كذا subscriber يهتم بيها.

Checkout -> database: order + outbox row in one transaction // (1)
Dispatcher -> broker -> email queue -> email worker          // (2)
                     -> analytics queue -> analytics worker

التكلفة: eventual consistency، و duplicates، وعمليات تشغيل أكتر. لازم pending state واضحة للمستخدم، ومراقبة عمر الرسائل. راجع topic 12 لتفاصيل CAP و consistency؛ messaging مش ضمان تلقائي للـ availability.

#RabbitMQ: المكونات والـ exchangesRabbitMQ components and exchange types

الـ producer بينشر للـ exchange ومع الرسالة routing key. الـ binding بيربط exchange بـ queue بقواعد routing، والـ consumer بيقرأ من queue. الـ connection غالباً طويلة العمر، وجواها channels أخف؛ الـ virtual host عزل منطقي بصلاحيات منفصلة.

الـ exchange مش هو مكان تخزين الـ backlog. الـ DLX كمان exchange، ولازم queue مربوطة بيه لو عايز تحتفظ بالفشل. الأنواع دي الأساسية في AMQP 0-9-1؛ RabbitMQ عنده أنواع وإضافات تانية.

exchange=orders (topic), routingKey=order.paid.eg
  binding=order.*.eg -> regional-orders
  binding=order.#    -> all-orders
  binding=order.*    -> no match (only two words)
exchange="", routingKey=email -> queue named email

#RabbitMQ.Client 7: النشر والـ confirmsRabbitMQ.Client 7: publishing and confirms

التطبيق اللي بينشر الرسالة اسمه publisher، وRabbitMQ هو broker اللي يستقبلها ويوجهها للـ queue. نجاح الكتابة على socket معناه إن البيانات خرجت من تطبيقك، مش إن الـ broker قبلها أو احتفظ بيها. Publisher confirm رد من الـ broker على نتيجة النشر؛ وهو مختلف عن consumer ack، إشارة من المستهلك بعد ما يخلص معالجة رسالة. المثال برنامج console بـ RabbitMQ.Client 7.x ويفترض broker محلي؛ production يحتاج TLS وصلاحيات محدودة وإعدادات خارج الكود.

using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory { HostName = "localhost" };
await using var connection = await factory.CreateConnectionAsync();
var options = new CreateChannelOptions(
    publisherConfirmationsEnabled: true,
    publisherConfirmationTrackingEnabled: true);                  // (1)
await using var channel = await connection.CreateChannelAsync(options);
await channel.QueueDeclareAsync(
    queue: "orders", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum"                               // (2)
    });
var properties = new BasicProperties
{
    Persistent = true,
    MessageId = Guid.NewGuid().ToString()
};
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
await channel.BasicPublishAsync(
    exchange: "", routingKey: "orders", mandatory: true,
    basicProperties: properties, body: Encoding.UTF8.GetBytes("order-created"),
    cancellationToken: timeout.Token);                           // (3)

#Acks و nack و prefetch و DLXAcknowledgments, prefetch and dead lettering

لما broker يسلّم رسالة للـ consumer، المستهلك يبعت ack (acknowledgment) بعد ما يخلص بنجاح علشان broker يعرف إنه يقدر يشيل التسليم من الانتظار. لو المستهلك حفظ التغيير في database وبعدين وقع قبل ack، broker ممكن يعيد الرسالة؛ عشان كده الـ handler لازم يتحمل التكرار من غير ما يكرر الأثر. Prefetch حد للرسائل المسلّمة اللي لسه ماعليهاش ack؛ مش هو نفسه عدد handlers اللي شغالة بالتوازي.

// Inside a BackgroundService: channel and HandleAsync are application collaborators.
// Provision the queue and a bound DLQ first; configure its DLX through broker policy.
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

await channel.BasicQosAsync(0, prefetchCount: 16, global: false); // (1)
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, delivery) =>
{
    byte[] body = delivery.Body.ToArray();                      // (2)
    try
    {
        await HandleAsync(body, stoppingToken);
    }
    catch (InvalidDataException)
    {
        await channel.BasicNackAsync(
            delivery.DeliveryTag, multiple: false, requeue: false); // (3)
        return;
    }
    await channel.BasicAckAsync(delivery.DeliveryTag, multiple: false); // (4)
};
await channel.BasicConsumeAsync("orders", autoAck: false, consumer: consumer);
await Task.Delay(Timeout.Infinite, stoppingToken);               // (5)

المثال بيفصل permanent failure بس. في transient failure لازم callback exception handling واضح، و bounded retry مع delay أو retry queue؛ ماتعتمدش على exception غير معالجة إنها هتعمل requeue لوحدها. إغلاق channel بيرجع unacked messages، لكن retry فوري بلا حد يعمل hot loop. راقب DLQ وحط replay workflow.

#Kafka: log و partitions و offsetsKafka: logs, partitions and offsets

لو عايز كذا نظام يعيد قراءة نفس التاريخ، queue بتمسح بعد ack مش دايماً النموذج المناسب. Kafka بيقسم topic لـ partitions، وكل partition سجل مترتب، وكل record ليه offset محلي. القراية مابتمسحش الحدث؛ retention بالوقت أو الحجم، و compaction إعداد مختلف يحتفظ بأحدث قيمة لكل key ضمن قواعده.

في consumer group التقليدية، كل partition لها consumer نشط واحد من نفس group. Groups مختلفة تقرأ نفس البيانات باستقلال. Commit قبل المعالجة ممكن يضيّع effect؛ بعد المعالجة ممكن يكررها بعد crash. الـ committed offset بيمثل موضع الاستكمال، مش إثبات إن كل الأنظمة الخارجية اتحدثت.

topic: orders
partition 0: offset 0 [order A v1]  offset 1 [order A v2] // (1)
partition 1: offset 0 [order B v1]
group invoicing: consumer X -> p0, consumer Y -> p1     // (2)
group analytics: consumer Z -> p0 and p1                // (3)

Producer retries وإعدادات idempotent producer والـ in-flight requests بتأثر على الحفاظ على الترتيب. Processing parallel جوه partition أو تخطي failed event ممكن يكسر ترتيب الـ effects.

#RabbitMQ و Kafka و Azure Service BusRabbitMQ, Kafka and Azure Service Bus

RabbitMQ عنده Streams كمان؛ المقارنة هنا للـ queues التقليدية. الاختيار حسب replay و routing و SLA وخبرة التشغيل والتكلفة، مش ادعاء إن واحد أسرع دائماً. Features زي duplicate detection في Service Bus ليها نطاق وزمن محدودين، مش ضمان exactly-once لكل business effect.

#Transactional outbox و inboxTransactional outbox and inbox

الحفظ وبعدين النشر عمليتين منفصلتين: crash بينهم يضيّع event، والنشر الأول ممكن يعلن عن order مااتحفظتش. خزّن business change و outbox row في نفس database transaction، وخلّي dispatcher ينشر بعدين. عند المستهلك، سجّل message ID مع الـ effect في transaction واحدة.

// EF Core fragments; entities and serializer are application code.
await using var tx = await db.Database.BeginTransactionAsync(ct);
db.Orders.Add(order);
db.Outbox.Add(new OutboxMessage(messageId, Serialize(order))); // (1)
await db.SaveChangesAsync(ct);
await tx.CommitAsync(ct);

// Dispatcher: claim a row safely before publishing; broker confirms required.
await PublishWithConfirmAsync(row.Id, row.Payload, ct);       // (2)
row.MarkSent();
await db.SaveChangesAsync(ct);

// Consumer uses a different DbContext and transaction.
await using var receiveTx = await db.Database.BeginTransactionAsync(ct);
db.Inbox.Add(new InboxMessage(consumerName, messageId));      // (3)
await db.SaveChangesAsync(ct);
await ApplyDatabaseEffectAsync(db, message, ct);             // (4)
await db.SaveChangesAsync(ct);
await receiveTx.CommitAsync(ct);
// Only now acknowledge the delivery.

#Delivery guarantees و idempotency والترتيبDelivery guarantees, idempotency and ordering

تعيين status ثابت ممكن يبقى idempotent، لكنه مش يحل stale event رجّع order من Shipped لـ Paid. استخدم sequence أو version لكل aggregate، وارفض gap أو اعمل buffer أو reconciliation حسب العقد.

في RabbitMQ، queue واحدة و single active consumer ومعالجة sequential تقلل إعادة الترتيب؛ requeue و retries والـ multi-producer محتاجين قواعد إضافية. في Kafka الترتيب داخل partition، مش بين partitions. لو outbox هتضمن ترتيب per order، ماتسمحش لـ row أحدث تتخطى row فاشلة لنفس order. مفيش global order مجاني.

#Sagas: choreography و orchestrationSagas: choreography and orchestration

حجز مخزون وتحصيل دفع في services مستقلة مش transaction محلية واحدة. Saga بتتبع سلسلة local transactions، ومع الفشل بتطلب compensations زي release و refund. Compensation فعل business جديد، مش rollback سحري ولا مسح لحقيقة الدفع.

Pending -> StockReserved -> PaymentCaptured -> Confirmed // (1)
StockReserved + PaymentRejected -> ReleasingStock -> Cancelled
PaymentCaptured + FulfilmentFailed -> Refunding -> Refunded // (2)

لو invariants لازم تتحقق فوراً، فكّر تخلّيها جوه boundary واحدة بدل إدخال saga. Saga مابتوفرش isolation؛ pending states وحجوزات مؤقتة بتحمي من سباقات العمليات.

#Resilience و messaging frameworksResilience and messaging frameworks

خدمة بطيئة ممكن تستهلك كل connections و threads المتاحة. Retry بعشوائية jitter يعالج failure مؤقت، و timeout يحد الانتظار، و circuit breaker يوقف النداء مؤقتاً بعد فشل متكرر، و bulkhead يعزل حصة concurrency لكل dependency. ماتعيدش validation failure، وماتكررّش charge غير آمن من غير idempotency contract.

// Package: Microsoft.Extensions.Http.Resilience; Polly v8-based integration.
using Microsoft.Extensions.Http.Resilience;

builder.Services.AddHttpClient("inventory")
    .AddStandardResilienceHandler(options =>
    {
        options.Retry.MaxRetryAttempts = 3;
        options.Retry.UseJitter = true;                   // (1)
        options.Retry.DisableForUnsafeHttpMethods();     // (2)
        options.AttemptTimeout.Timeout = TimeSpan.FromSeconds(2);
        options.TotalRequestTimeout.Timeout = TimeSpan.FromSeconds(10); // (3)
    });

MassTransit و Wolverine بيضيفوا consumer pipelines و retries و durable messaging و saga support حسب التكامل المختار. لازم تضبط persistence والـ outbox والـ error handling صراحة؛ in-memory outbox مش بديل durable transaction. HTTP handler ده لحماية outgoing HTTP، مش consumer retry policy.