🚀 Competing Consumers Pattern باستخدام Redis Streams و .NET Core

لما Worker واحد يقولك: «يا جماعة الطلبات كتير عليّا!» 😅

تخيل إن عندك E-Commerce System، والعميل بيعمل Order جديد.

الـ API تعمل شغلها الطبيعي:

POST /orders

تحفظ الـ Order في الـ Database، وبعد كده محتاج تعمل شوية عمليات تقيلة في الخلفية:

💳 تجهيز عملية الدفع.
📦 تجهيز الـ Order للشحن.
🧾 إنشاء Invoice.
📧 إرسال Confirmation Email.
📊 تحديث Analytics.

طبعًا مش منطقي تخلي العميل واقف مستني كل ده يحصل قبل ما الـ API ترجع Response.

فأول فكرة ممكن تيجي في دماغنا:

Task.Run(async () =>
{
   await ProcessOrderAsync(order);
});

خلصنا يا باشا 😎

للأسف… لأ. 😄

لأن أول ما السيستم يكبر، وتبدأ تشغل أكتر من Server، وتحصل Restart أو Deployment أو Crash، هتكتشف إننا محتاجين حاجة أقوى بكتير.

وهنا بيدخل معانا:

⚡ Competing Consumers Pattern

ومعاه:

🔴 Redis Streams + Consumer Groups

🎯 الأول: يعني إيه Competing Consumers؟

الفكرة بسيطة جدًا.

بدل ما يكون عندك Worker واحد بيقرأ الـ Messages:

             ┌───────────────┐
Messages ───▶│   Worker #1   │
             └───────────────┘

ممكن يكون عندك 3 أو 5 أو 20 Workers:

                       ┌──────────────┐
                  ┌──▶│ Consumer #1  │
                  │   └──────────────┘
                  │
Messages ──────────┼──▶┌──────────────┐
                  │   │ Consumer #2  │
                  │   └──────────────┘
                  │
                  └──▶┌──────────────┐
                      │ Consumer #3  │
                      └──────────────┘

لكن المهم هنا:

التلاتة مش بيعالجوا نفس الـ Message.

هم بيتنافسوا Compete على الشغل.

يعني لو عندنا:

Order-101
Order-102
Order-103
Order-104
Order-105
Order-106

ممكن التوزيع يحصل كده:

🟢 Consumer-A
 ├── Order-101
 └── Order-104

🔵 Consumer-B
 ├── Order-102
 └── Order-105

🟣 Consumer-C
 ├── Order-103
 └── Order-106

كل Consumer خد جزء من الشغل.

وده بالضبط مفهوم:

Competing Consumers

Redis Consumer Groups معمولة عشان السيناريو ده تحديدًا؛ داخل الـ Group الواحد، الـ consumers بيتقاسموا الـ entries بدل ما كل واحد يستلم نسخة منها.

🔴 طيب إيه Redis Streams أصلًا؟

Redis Stream تقدر تتخيله كأنه:

Append-Only Event Log

يعني بدل ما تحط قيمة عادية:

key → value

عندك سلسلة Messages:

orders:stream

┌───────────────────────────────┐
│ 1750000010000-0  Order #101  │
├───────────────────────────────┤
│ 1750000010200-0  Order #102  │
├───────────────────────────────┤
│ 1750000010300-0  Order #103  │
├───────────────────────────────┤
│ 1750000010400-0  Order #104  │
└───────────────────────────────┘

وكل Entry ليها ID مرتب زمنيًا.

الـ Producer يضيف باستخدام:

XADD

والـ Consumers تقرأ باستخدام:

XREADGROUP

وبعد ما الـ Consumer يخلص:

XACK

Redis Streams بتحتفظ بالـ entries وتوفر replay وconsumer tracking وacknowledgment، وده مختلف عن Redis Pub/Sub اللي بطبيعته fire-and-forget ومفيهوش history للرسائل القديمة.

🧠 أهم جزء: Consumer Group

خلينا نقول إن عندنا Stream اسمه:

orders:stream

وعندنا Consumer Group:

order-processors

وجواه:

order-worker-01
order-worker-02
order-worker-03

الصورة تبقى:

                    🔴 Redis
             ┌──────────────────┐
             │  orders:stream   │
             │                  │
Producer ────▶│ Order-101        │
             │ Order-102        │
             │ Order-103        │
             │ Order-104        │
             │ Order-105        │
             └────────┬─────────┘
                      │
                      ▼
            🟠 Consumer Group
             order-processors
                      │
         ┌────────────┼────────────┐
         ▼            ▼            ▼
    🟢 Worker-1   🔵 Worker-2   🟣 Worker-3
      101            102            103
      104            105            ...

Redis هو اللي بيمسك:

مين استلم إيه؟
ومين خلص؟
ومين استلم Message ولسه ماعملهاش ACK؟

وده بيتم عن طريق حاجة اسمها:

📋 Pending Entries List — PEL

📦 السيناريو العملي

هنبني System بسيط:

Orders API
    │
    │ OrderCreated
    ▼
Redis Stream
"orders:created"
    │
    ▼
Consumer Group
"order-processing"
    │
 ┌───┼─────────────┐
 ▼   ▼             ▼
W1   W2            W3

كل Worker ممكن يكون موجود على Server مختلف.

مثلاً:

VM-01
└── OrderWorker

VM-02
└── OrderWorker

VM-03
└── OrderWorker

أو Kubernetes:

order-worker Deployment

Replica #1
Replica #2
Replica #3
Replica #4
Replica #5

كلهم يشتركوا في نفس:

Consumer Group = order-processing

لكن كل Instance لازم يكون ليه:

Unique Consumer Name

🛠️ إنشاء المشروع

هنستخدم:

.NET 8+
StackExchange.Redis
Redis Streams
BackgroundService

ممكن تعمل Worker:

dotnet new worker -n OrderProcessor.Worker

وتضيف Redis Client:

dotnet add package StackExchange.Redis

StackExchange.Redis هو الـ Redis client الشائع في .NET، وRedis نفسه عنده أمثلة رسمية لاستخدام Streams من C#.

🔌 الاتصال بـ Redis

في:

Program.cs

نعمل:

using StackExchange.Redis;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddSingleton<IConnectionMultiplexer>(_ =>
{
   return ConnectionMultiplexer.Connect(
       builder.Configuration.GetConnectionString("Redis")!
   );
});

builder.Services.AddSingleton(sp =>
{
   var redis = sp.GetRequiredService<IConnectionMultiplexer>();
   return redis.GetDatabase();
});

builder.Services.AddHostedService<OrderConsumerWorker>();

var host = builder.Build();

host.Run();

وفي:

appsettings.json

نحط:

{
 "ConnectionStrings": {
   "Redis": "localhost:6379"
 }
}

Microsoft بتوفر BackgroundService وIHostedService تحديدًا للـ long-running background workloads، وبتقدر تسجل الـ worker باستخدام AddHostedService.

📤 Producer: إضافة Order للـ Stream

نفترض إن عندنا Event:

public record OrderCreatedEvent(
   Guid EventId,
   Guid OrderId,
   Guid CustomerId,
   decimal Total,
   DateTimeOffset CreatedAt
);

نقدر ننشره:

using System.Text.Json;
using StackExchange.Redis;

public class OrderEventPublisher
{
   private const string StreamName = "orders:created";

   private readonly IDatabase _redis;

   public OrderEventPublisher(IDatabase redis)
   {
       _redis = redis;
   }

   public async Task PublishAsync(
       OrderCreatedEvent order)
   {
       var json = JsonSerializer.Serialize(order);

       var values = new NameValueEntry[]
       {
           new("eventId", order.EventId.ToString()),
           new("orderId", order.OrderId.ToString()),
           new("payload", json),
           new(
               "createdAt",
               order.CreatedAt.ToUnixTimeMilliseconds()
           )
       };

       await _redis.StreamAddAsync(
           StreamName,
           values
       );
   }
}

دي تقريبًا بتترجم في Redis إلى:

XADD orders:created *
   eventId ...
   orderId ...
   payload ...

Redis يعمل Entry زي:

orders:created

1750000001234-0
│
├── eventId   = "..."
├── orderId   = "..."
├── payload   = "{...}"
└── createdAt = "..."

💡 طب ليه Stream بدل List؟

لأن Redis Stream مش مجرد:

Queue

هو عنده مفهوم:

Message History
Consumer Groups
Pending Messages
Acknowledgment
Replay
Claiming

يعني Redis عارف إن:

Order-101

اتسلم لـ:

worker-02

لكن لسه:

❌ Not Acknowledged

ودي معلومة شديدة الأهمية وقت الأعطال.

👥 إنشاء Consumer Group

قبل القراءة محتاجين Group:

order-processing

في StackExchange.Redis:

private async Task EnsureGroupExistsAsync()
{
   try
   {
       await _redis.StreamCreateConsumerGroupAsync(
           "orders:created",
           "order-processing",
           "0-0",
           createStream: true
       );
   }
   catch (RedisServerException ex)
       when (ex.Message.Contains("BUSYGROUP"))
   {
       // Group already exists.
   }
}

0-0 معناها:

لما الـ Group يتعمل، ابدأ من أول الـ Stream.

لو استخدمت:

معناه:

ابدأ من الرسائل الجديدة اللي تيجي بعد إنشاء الـ Group.

🧵 نعمل الـ Consumer Worker

public class OrderConsumerWorker : BackgroundService
{
   private const string StreamName =
       "orders:created";

   private const string GroupName =
       "order-processing";

   private readonly IDatabase _redis;

   private readonly ILogger<OrderConsumerWorker>
       _logger;

   private readonly string _consumerName;

   public OrderConsumerWorker(
       IDatabase redis,
       ILogger<OrderConsumerWorker> logger)
   {
       _redis = redis;
       _logger = logger;

       _consumerName =
           $"{Environment.MachineName}-" +
           $"{Guid.NewGuid():N}";
   }

   protected override async Task ExecuteAsync(
       CancellationToken stoppingToken)
   {
       await EnsureGroupExistsAsync();

       while (!stoppingToken.IsCancellationRequested)
       {
           var entries =
               await _redis.StreamReadGroupAsync(
                   StreamName,
                   GroupName,
                   _consumerName,
                   ">",
                   count: 10
               );

           if (entries.Length == 0)
           {
               await Task.Delay(
                   500,
                   stoppingToken
               );

               continue;
           }

           foreach (var entry in entries)
           {
               await ProcessMessageAsync(
                   entry,
                   stoppingToken
               );
           }
       }
   }
}

ركز معايا في:

">"

دي معناها:

هات Messages جديدة لم يتم تسليمها لأي Consumer داخل الـ Group.

وده بالضبط أساس الـ competing behavior في Redis Consumer Groups.

🔥 وهنا السحر بيحصل

لو عندنا 3 Instances من التطبيق:

Consumer Name:

SERVER-A-8fa23...
SERVER-B-24fd1...
SERVER-C-91ab4...

كلهم بينفذوا:

XREADGROUP
GROUP order-processing

Redis ممكن يوزع:

                    orders:created
                        │
        ┌───────────────┼───────────────┐
        │               │               │
        ▼               ▼               ▼

🟢 SERVER-A        🔵 SERVER-B        🟣 SERVER-C

Order-101          Order-102          Order-103
Order-104          Order-106          Order-105
Order-109          Order-107          Order-108

إذن لو الضغط زاد:

3 Workers

خليهم:

10 Workers

مش محتاج تغير الـ Producer.

ومش محتاج تغير الـ API.

ومش محتاج تعمل Distribution Logic بنفسك.

وده:

🚀 Horizontal Scaling

✅ ACK: اللحظة المهمة جدًا

بعد ما Worker يستلم Message:

Redis
│
│ Order-101
▼
Worker-2

Redis بيعتبرها:

Pending

مش Finished.

يعني:

orders:created

Order-101
   │
   ▼
Consumer: worker-2
Status: PENDING ⚠️

بعد ما نخلص الـ Business Logic:

await ProcessOrderAsync(order);

await _redis.StreamAcknowledgeAsync(
   StreamName,
   GroupName,
   entry.Id
);

ساعتها فقط:

PENDING
  │
  │ XACK
  ▼
✅ Processed

Redis بيوفر XACK عشان يشيل الرسالة من Pending Entries List بعد نجاح المعالجة.

🧩 ProcessMessageAsync

مثلاً:

private async Task ProcessMessageAsync(
   StreamEntry entry,
   CancellationToken cancellationToken)
{
   try
   {
       var payload =
           entry.Values
               .First(x => x.Name == "payload")
               .Value
               .ToString();

       var order =
           JsonSerializer.Deserialize<OrderCreatedEvent>(
               payload
           );

       if (order is null)
           throw new InvalidOperationException(
               "Invalid order message."
           );

       _logger.LogInformation(
           "Consumer {Consumer} processing Order {OrderId}",
           _consumerName,
           order.OrderId
       );

       await ProcessOrderAsync(
           order,
           cancellationToken
       );

       await _redis.StreamAcknowledgeAsync(
           StreamName,
           GroupName,
           entry.Id
       );

       _logger.LogInformation(
           "Order {OrderId} completed.",
           order.OrderId
       );
   }
   catch (Exception ex)
   {
       _logger.LogError(
           ex,
           "Failed processing message {MessageId}",
           entry.Id
       );

       // مهم:
       // ما نعملش ACK.
   }
}

وهنا خد بالك من القرار المهم:

Success
  ↓
XACK ✅

Failure
  ↓
NO ACK ❌

💥 طب لو Worker وقع في نص التنفيذ؟

هنا بقى Redis Streams بتظهر قوتها.

تخيل:

Redis
│
│ Order-501
▼
Worker-A
│
├── Received ✅
│
├── Started processing...
│
💥 CRASH

Worker-A ماعملش:

XACK

إذن Redis شايف:

Order-501

Owner:
Worker-A

State:
PENDING

الرسالة مش اختفت.

وده جزء أساسي من reliability model بتاع Consumer Groups؛ Redis يقدر يعرض الـ pending entries، والرسائل اللي فضلت بدون ACK ممكن تتعمل لها claim بواسطة Consumer سليم باستخدام XCLAIM أو XAUTOCLAIM.

🛟 XAUTOCLAIM

مثلاً:

Worker-A
💀 Dead

Order-501
⏳ Pending for 60 seconds

Worker-B يقدر يقول:

أنا هاخد الرسائل اللي أصحابها وقعوا.

باستخدام:

XAUTOCLAIM

الصورة:

              Pending Entries
                   │
            Order-501
                   │
                   ▼
            💀 Worker-A
                   │
            idle > 60 sec
                   │
                   ▼
             XAUTOCLAIM
                   │
                   ▼
            🟢 Worker-B
                   │
                   ▼
             Retry Order-501

StackExchange.Redis بيدعم StreamAutoClaimAsync، وبالتالي تقدر تعمل Recovery Worker أو periodic recovery loop للـ abandoned messages.

🔁 إذن Redis Streams بيديني At-Least-Once Delivery

ودي نقطة مهمة جدًا.

مش:

Exactly Once

ولكن تقريبًا:

Message delivered
    │
    ▼
Process
    │
    ├──── Success ────▶ ACK
    │
    └──── Crash ──────▶ Retry

وده معناه إن الرسالة ممكن تتعالج:

مرة

أو في بعض حالات الأعطال:

أكتر من مرة

ومن هنا تيجي أهمية:

🛡️ Idempotency

مثلاً لو Order:

OrderId = 500

اتعالج مرتين، ماينفعش تعمل:

Charge Customer
Charge Customer again 😱

لازم الـ processing layer تبقى مصممة بحيث إعادة نفس الـ message ما تعملش side effects مكررة.

🧠 دلوقتي السؤال المهم:

إيه الفرق عن Task.Run؟

نفترض إن الـ Controller فيه:

[HttpPost]
public async Task<IActionResult> CreateOrder()
{
   var order = await CreateOrderAsync();

   _ = Task.Run(async () =>
   {
       await ProcessOrderAsync(order);
   });

   return Ok();
}

ظاهريًا الموضوع جميل:

HTTP Request
  │
  ├── Create Order
  │
  └── Task.Run()
          │
          ▼
    Background Work

لكن عندك مشاكل معمارية كبيرة.

💣 المشكلة الأولى: Process Crash

مع:

Task.Run

الشغل موجود في Memory بتاعة:

ASP.NET Process

يعني:

Order
 │
 ▼
Task.Run()
 │
 ▼
RAM

لو حصل:

Application Restart
IIS Recycle
Pod Restart
Deployment
VM Crash
Process Kill

فالـ Task ممكن ببساطة:

💨 تختفي.

مفيش Broker خارجي فاكر:

كان فيه Order لسه محتاج يتعالج.

Microsoft نفسها بتفرق بين arbitrary background threads وبين hosted services المرتبطة بعمر التطبيق، وبتوصي باستخدام Hosted Services للـ long-running background workloads بدل الاعتماد على thread عشوائي غير مُدار.

🔴 Redis Streams

هنا الوضع مختلف:

Application
  │
  │ XADD
  ▼
┌────────────────────┐
│      Redis         │
│                    │
│ Order-101          │
│ Order-102          │
│ Order-103          │
└────────────────────┘

الـ Application نفسها ممكن تموت:

Application 💀

لكن الـ Stream موجود خارج الـ process.

Worker جديد يطلع:

Worker-B 🟢

ويرجع يكمل.

وده اللي ممكن نسميه هنا:

Persistence / Recoverability / Rebindability

يعني الـ Work مش مربوط بعمر الـ Process اللي أنشأه.

🧵 طيب والـ Channel<T>؟

.NET عنده:

Channel<T>

ودي حاجة ممتازة جدًا.

مثلاً:

var channel =
  Channel.CreateBounded<Order>(1000);

Producer:

await channel.Writer.WriteAsync(order);

Consumer:

await foreach (
  var order in channel.Reader.ReadAllAsync())
{
  await ProcessAsync(order);
}

ده ممتاز لو كل الشغل:

داخل نفس الـ Process

🟢 Architecture مع Channel

┌─────────────────────────────┐
│       ASP.NET Process       │
│                             │
│ API                         │
│  │                          │
│  ▼                          │
│ Channel<Order>              │
│  │                          │
│  ├── Worker 1               │
│  ├── Worker 2               │
│  └── Worker 3               │
│                             │
└─────────────────────────────┘

جميل جدًا.

لكن كله موجود داخل:

Process واحد.

🔴 Architecture مع Redis Streams

┌──────────────┐
│ Orders API   │
│ Server A     │
└──────┬───────┘
      │
      │ XADD
      ▼
╔════════════════════════╗
║      REDIS STREAM      ║
║                        ║
║ orders:created         ║
╚═══════════╤════════════╝
           │
           │ Consumer Group
           │
     ┌─────┼─────┐
     │     │     │
     ▼     ▼     ▼

┌────────┐ ┌────────┐ ┌────────┐
│Worker A│ │Worker B│ │Worker C│
│Server 1│ │Server 2│ │Server 3│
└────────┘ └────────┘ └────────┘

وهنا إحنا دخلنا عالم:

🌍 Distributed Processing

⚖️ Redis Streams vs Channel vs Task.Run

الخاصية	Task.Run	Channel<T>	Redis Streams
Background Processing	✅	✅	✅
داخل نفس Process	✅	✅	❌ مش لازم
Multi-Server	❌	❌	✅
Message History	❌	❌	✅
Replay	❌	❌	✅
Consumer Groups	❌	In-Process فقط	✅
Explicit ACK	❌	❌	✅
Pending Messages	❌	❌	✅
Recovery بعد Worker Crash	ضعيف	ضعيف	✅
Horizontal Scaling	محدود	داخل Process	✅
Distributed Consumers	❌	❌	✅
Retry Infrastructure	Manual	Manual	✅ قابل للبناء فوق PEL
Survives App Restart	❌	❌	✅
مناسب لـ durable jobs	❌	❌	✅ مع إعداد Redis مناسب

🎯 إمتى أستخدم Channel؟

أنا مش بقول:

Redis Streams replaces Channel

لأ خالص.

Channel<T> ممتاز لما المشكلة:

In-Process Concurrency

مثلاً:

Image Processing Pipeline
داخل Service واحد

أو:

Buffer بين Producer وConsumer
داخل نفس Application Instance

وقتها إدخال Redis ممكن يكون Overengineering.

🎯 إمتى Redis Streams؟

Redis Streams تبدأ تبقى جذابة لما تقول:

الشغل مهم ومينفعش يضيع.

أو:

أنا محتاج أكتر من Server يعالج.

أو:

لو Worker وقع حد تاني يكمل.

أو:

عايز أزود Workers وقت الضغط.

أو:

عايز أعرف مين استلم الرسالة ولسه ماخلصهاش.

ساعتها إحنا خرجنا من:

Background Programming

ودخلنا في:

Distributed Messaging Architecture

🧠 نقطة معمارية مهمة جدًا:

Consumer ≠ Consumer Group

دي من أكتر الحاجات اللي بتتلخبط.

نفترض إن الـ Order لازم يحصل له:

Payment
Notification
Analytics

لو حطيت التلاتة Consumers في:

Consumer Group واحد

هيحصل إيه؟

Order-100
  │
  ▼
Consumer Group
  │
  ├──── Payment Worker
  ├──── Notification Worker
  └──── Analytics Worker

Redis هيختار واحد فقط منهم يستلم الـ Message.

وده غلط.

لأننا محتاجين التلات خدمات تشوف الـ Order.

✅ الحل

نعمل:

orders:created

لكن 3 Groups:

payment-processors

notification-processors

analytics-processors

فتبقى الصورة:

                    🔴 orders:created
                          │
         ┌────────────────┼────────────────┐
         │                │                │
         ▼                ▼                ▼

🟢 payment-group   🔵 notification-group   🟣 analytics-group

     │                    │                    │
┌────┼────┐          ┌────┼────┐          ┌────┼────┐
▼    ▼    ▼          ▼    ▼    ▼          ▼    ▼    ▼
P1   P2   P3         N1   N2   N3         A1   A2   A3

كل Group يشوف الـ Stream بشكل مستقل، لكن جوّه كل Group الـ Workers ينافسوا بعض.

دي واحدة من أقوى الأفكار في Redis Streams. Redis بيسمح بأكتر من consumer group لنفس الـ stream، وكل Group عنده cursor وتتبع مستقل.

⚡ مثال Scaling حقيقي

الساعة 11 الصبح:

Orders/sec = 50

عندك:

2 Workers

كويسين جدًا.

الساعة 8 مساءً Black Friday:

Orders/sec = 5000

تقدر تعمل:

kubectl scale deployment order-worker
--replicas=20

فتتحول:

Redis Stream
   │
┌───┼───┐
▼   ▼   ▼
W1  W2  W3

إلى:

Redis Stream
│
├── W01
├── W02
├── W03
├── W04
├── W05
├── ...
└── W20

من غير ما الـ Producer يعرف أصلاً إنك زودت Workers.

ودي نقطة شديدة الأهمية:

Producer و Consumers Decoupled

🔥 Backpressure بشكل طبيعي

تخيل:

Producer Rate
1000 msg/sec

لكن Workers يقدروا يعالجوا:

600 msg/sec

مع Task.Run ممكن تبدأ تولع Thread Pool 😅:

1000
2000
5000
10000 Tasks...

لكن في Stream:

Producer
1000 msg/sec
    │
    ▼
Redis Stream
████████████████
    │
    ▼
Consumers
600 msg/sec

الفرق:

400 Message/sec

يتراكم في الـ Stream بدل ما يبقى آلاف الـ Tasks العشوائية داخل Memory بتاعة الـ API.

وده يسمح لك تراقب:

Consumer Lag
Pending Messages
Processing Rate

وتاخد قرار:

Scale Out 🚀

Redis بيوفر group state وpending information وأدوات زي XPENDING تساعدك تراقب الرسائل اللي لسه تحت المعالجة.

⚠️ ACK قبل ولا بعد الـ Processing؟

ممنوع تعمل:

Receive
↓
ACK
↓
Process

ليه؟

لأن السيناريو ده ممكن يحصل:

Receive Order
    │
    ▼
ACK ✅
    │
    ▼
Process
    │
    ▼
💥 CRASH

Redis فاكر إنك خلصت.

لكن الحقيقة:

Order ضاع.

الصح:

Receive
 │
 ▼
Process
 │
 ▼
Business Transaction
 │
 ▼
Success
 │
 ▼
XACK ✅

🛡️ Transactional Outbox مهم برضه

Redis Streams بتحل لك مشكلة:

Consumer Reliability

لكن في Producer فيه مشكلة مختلفة.

نفترض إن Orders API تعمل:

INSERT Order INTO SQL

وبعدها:

XADD orders:created

ممكن يحصل:

SQL INSERT ✅

      ↓

Application Crash 💥

      ↓

XADD ❌

الـ Order اتحفظ.

لكن الـ Event ما اتبعتش.

ودي اسمها:

Dual Write Problem

لو الـ Event Critical، الحل الأشهر معماريًا يبقى:

SQL Transaction
   │
   ├── Insert Order
   └── Insert OutboxMessage
            │
            ▼
      Outbox Publisher
            │
            ▼
        Redis Stream

يعني Redis Streams مش بديل عن Transactional Outbox؛ الاتنين بيحلوا مشكلتين مختلفين.

🧬 Architecture النهائية

                      👤 Customer
                          │
                          ▼
                 ┌────────────────┐
                 │   Orders API   │
                 └────────┬───────┘
                          │
                          ▼
                   SQL Transaction
                          │
               ┌─────────┴────────┐
               ▼                  ▼
           🛒 Orders          📤 Outbox
                                  │
                                  ▼
                        Outbox Publisher
                                  │
                                  │ XADD
                                  ▼

                 ╔══════════════════════╗
                 ║     REDIS STREAM     ║
                 ║                      ║
                 ║    orders:created    ║
                 ╚══════════╤═══════════╝
                            │
                Consumer Group:
                order-processing
                            │
             ┌──────────────┼──────────────┐
             │              │              │
             ▼              ▼              ▼
       🟢 Worker-01    🔵 Worker-02    🟣 Worker-03
             │              │              │
             ▼              ▼              ▼
         Process          Process         Process
             │              │              │
             └──────┬───────┴───────┬─────┘
                    │               │
                Success          Failure
                    │               │
                    ▼               ▼
                XACK ✅        Pending ⚠️
                                    │
                                    ▼
                                Retry /
                              XAUTOCLAIM

دي Architecture قريبة جدًا من اللي ممكن تستخدمه فعلاً في Production.

🧠 الخلاصة المعمارية

الـ Competing Consumers Pattern مش مجرد:

نشغل شوية Threads.

الفكرة الحقيقية:

                 Workload
                   │
                   ▼
         Shared Message Source
                   │
                   ▼
           Consumer Group
                   │
      ┌────────────┼────────────┐
      ▼            ▼            ▼
   Worker A     Worker B     Worker C

وكل Worker:

Stateless قدر الإمكان

وبالتالي تقدر:

Scale Out
Scale In
Restart
Deploy
Recover
Retry

من غير ما الـ Producer يكون مرتبط بعدد الـ Workers.

🥊 Task.Run vs Channel vs Redis Streams في جملة واحدة

`Task.Run`

"اعمل الحاجة دي في الخلفية جوه نفس الـ Process."

`Channel<T>`

"اعمل لي Producer/Consumer Queue محترمة،
بس جوه نفس الـ Process."

`Redis Streams + Consumer Groups`

"اعمل لي Distributed Work Queue / Event Stream
تقدر Workers على Servers مختلفة تتنافس عليها،
مع Tracking وACK وRecovery."

ودي هي النقلة المعمارية الحقيقية. 🚀

🎯 قاعدة سهلة تحفظها

لو الشغل:

Cheap + Local + Loss Acceptable

فكر في:

Task / Channel

لكن لو الشغل:

Important
+
Distributed
+
Must Survive App Restarts
+
Needs Retry
+
Needs Horizontal Scaling

ابدأ تفكر في:

Redis Streams
RabbitMQ
Azure Service Bus
Kafka

حسب طبيعة المشكلة.

Redis Streams اختيار لطيف جدًا خصوصًا لما يكون Redis موجود بالفعل في الـ infrastructure ومحتاج Streaming/queue-like capabilities بدون إدخال منصة أكبر من احتياجك. Redis نفسها بتقدم Streams كحل للـ ordered event processing والconsumer groups والreplay والshort/moderate retention workloads.

⚠️ Challenges لازم تبقى في حسابك

وأنت بتحول التصميم ده لـ Production، حط في Architecture Checklist:

☠️ Poison Messages
🔁 Retry Strategy
🛡️ Idempotency
🔢 Message Ordering
💀 Dead Letter Queue / Dead Letter Stream
📋 Pending Entries Management
🛟 XAUTOCLAIM / Abandoned Consumers
📈 Consumer Lag Monitoring
📦 Stream Retention / Trimming
💾 Redis Persistence Configuration
🔐 Security & Authentication
🔄 Graceful Shutdown
🧬 Message Schema Versioning
🆔 Unique Consumer Naming
⚖️ Load Distribution
🧾 Transactional Outbox
🔭 Observability & Distributed Tracing

🚀 الخلاصة

أهم حاجة تطلع بيها من المقال إن:

Competing Consumers

مش مجرد Optimization عشان نخلي الـ processing أسرع.

هو Pattern بيغير طريقة تفكيرك من:

Application
 │
 └── Background Task

إلى:

                   Distributed System

Producer
 │
 ▼
Durable / Recoverable Work
 │
 ▼
Consumer Group
 │
 ├── Consumer #1
 ├── Consumer #2
 ├── Consumer #3
 └── Consumer #N

فبدل ما الـ API تقول:

"أنا هعمل الـ Order Processing بنفسي في الخلفية."

بتقول:

"أنا هسجل إن فيه Work محتاج يتعمل، وأسيب Pool من الـ Workers يتنافس عليه."

وده بيديك:

🚀 Scalability

🛡️ Reliability

🔁 Retries

💾 Persistence / Recoverability

🌍 Distributed Processing

⚡ Loose Coupling

وده بالضبط السبب اللي بيخلي Redis Streams + Consumer Groups اختيار قوي جدًا لتنفيذ Competing Consumers Pattern في .NET.