کانال‌های رویداد (Event Channels) در معماری مبتنی بر رویداد (EDA) چیست؟

کانال‌های رویداد مسیرهایی هستند که از طریق آن‌ها رویدادها بین سرویس‌های مختلف منتقل می‌شوند. می‌توان این کانال‌ها را مشابه خطوط لوله (Pipeline) یا صف‌های پیام (Queue) در نظر گرفت که وظیفه انتقال رویدادها از یک سرویس تولیدکننده (Producer) به یک یا چند سرویس مصرف‌کننده (Consumer) را بر عهده دارند.

تعریف و نقش کانال‌های رویداد

زمانی که یک Event Generator (مانند یک سرویس پرداخت یا ثبت سفارش) رویدادی را ایجاد می‌کند، این رویداد باید از طریق یک کانال مناسب به مقصد برسد. این کانال می‌تواند به صورت صف (Queue)، موضوع (Topic) یا جریان داده (Stream) پیاده‌سازی شود.

اهمیت کانال‌های رویداد

• انتقال ناهمزمان (Asynchronous): رویدادها بدون نیاز به پاسخ فوری پردازش می‌شوند.
• انعطاف‌پذیری (Resilience): در صورت عدم دسترسی موقت یک سرویس، رویدادها در کانال باقی می‌مانند تا زمانی که سرویس مجدداً فعال شود.
• مقیاس‌پذیری (Scalability): امکان اتصال چندین سرویس به یک کانال و پردازش موازی رویدادها وجود دارد.
• اتصال سست (Loose Coupling): سرویس‌های تولیدکننده و مصرف‌کننده از یکدیگر مستقل هستند و وابستگی مستقیم ندارند.

انواع کانال‌های رویداد

1. صف (Queue)

• هر رویداد فقط توسط یک مصرف‌کننده پردازش می‌شود.
• مثال: صف OrderQueue برای پردازش سفارش‌ها.
• کاربرد: مناسب برای سناریوهایی که پردازش رویداد باید دقیقاً یک بار انجام شود (مانند پردازش پرداخت).

2. موضوع (Topic)

• هر رویداد می‌تواند توسط چندین مصرف‌کننده دریافت شود.
• مثال: موضوع OrderPlacedTopic که هم توسط سرویس ارسال و هم سرویس فاکتورخوانی استفاده می‌شود.
• کاربرد: مناسب برای مواردی که چندین سرویس باید از یک رویداد مطلع شوند.

3. جریان داده (Stream)

• رویدادها به صورت یک جریان پیوسته ذخیره و پردازش می‌شوند.
• مثال: PaymentStream که تاریخچه تراکنش‌ها را نگهداری می‌کند.
• کاربرد: مناسب برای پردازش بلادرنگ (Real-time) و تحلیل داده‌های حجیم.

پیاده‌سازی در .NET (با RabbitMQ یا Kafka)

1. تعریف صف برای پردازش سفارش‌ها

public class OrderPlacedHandler
{
    private readonly IEventBus _eventBus;

    public OrderPlacedHandler(IEventBus eventBus)
    {
        _eventBus = eventBus;
    }

    public void HandleOrderPlaced(OrderPlacedEvent orderEvent)
    {
        Console.WriteLine($&quotProcessing order {orderEvent.OrderId}&quot);
        _eventBus.PublishToQueue(&quotOrderQueue&quot, orderEvent);
    }
}

2. تعریف موضوع برای اطلاع‌رسانی به چند سرویس

public class OrderNotifier
{
    private readonly IEventBus _eventBus;

    public OrderNotifier(IEventBus eventBus)
    {
        _eventBus = eventBus;
    }

    public void NotifyServices(OrderPlacedEvent orderEvent)
    {
        _eventBus.PublishToTopic(&quotOrderPlacedTopic&quot, orderEvent);
    }
}

نکات کلیدی در طراحی کانال‌های رویداد

چند مصرف‌کننده (Multiple Consumers): در مدل Topic، چندین سرویس می‌توانند یک رویداد را دریافت کنند.
صف مرده (Dead Letter Queue - DLQ): رویدادهای پردازش‌نشده به یک صف جداگانه منتقل می‌شوند تا بعداً بررسی شوند.
حفظ ترتیب رویدادها (Event Ordering): در برخی سیستم‌ها مانند Kafka، ترتیب رویدادها حفظ می‌شود.

کاربردهای واقعی

• پردازش پرداخت: ارسال رویداد PaymentProcessed به سرویس‌های حسابداری و انبارداری.
• مانیتورینگ: ارسال رویداد ErrorOccurred به سرویس‌های گزارش‌گیری.
• اینترنت اشیا (IoT): ارسال داده‌های سنسورها به یک جریان برای تحلیل بلادرنگ.

چالش‌های اضافه بار در سیستم‌های مبتنی بر رویداد

زمانی که حجم رویدادها از حد معمول فراتر رود، سیستم با مشکلات متعددی مواجه می‌شود:

• افزایش تأخیر در پردازش (Latency)
• انباشت پیام‌ها در صف‌ها (Queue Overflow)
• از کار افتادن مصرف‌کننده‌ها (Consumers Crash)
• از دست رفتن پیام‌ها (Message Loss)

استراتژی‌های مدیریت اضافه بار

۱. کنترل فشار پردازش (Backpressure Handling)

مشکل: عدم توانایی مصرف‌کننده‌ها در پردازش رویدادها با سرعت مناسب

راهکارها:

• محدودسازی نرخ تولید (Rate Limiting): کنترل تعداد رویدادهای تولید شده در واحد زمان
• کنترل جریان در Message Broker: تنظیم حداکثر پیام‌های در حال پردازش
• الگوی قطع ارتباط (Circuit Breaker): توقف موقت پردازش در صورت overload

مثال پیاده‌سازی در RabbitMQ:

channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);

۲. توازن بار بین مصرف‌کننده‌ها (Load Balancing)

مشکل: توزیع ناعادلانه بار پردازش بین مصرف‌کننده‌ها

راهکارها:

• افزایش افقی مصرف‌کننده‌ها (Horizontal Scaling)
• استفاده از Consumer Groups در Kafka
• پیاده سازی Autoscaling در Kubernetes

مثال پیاده‌سازی در Kafka:

var config = new ConsumerConfig
{
    BootstrapServers = &quotlocalhost:9092&quot,
    GroupId = &quotorder-processing-group&quot,
    EnableAutoCommit = false
};

using var consumer = new ConsumerBuilder<Ignore, string>(config).Build();
consumer.Subscribe(&quotorder-events&quot);
var consumeResult = consumer.Consume();
Console.WriteLine($&quotProcessing event: {consumeResult.Message.Value}&quot);
consumer.Commit(consumeResult);

۳. اولویت‌بندی پیام‌ها (Message Prioritization)

مشکل: تأخیر در پردازش رویدادهای حیاتی

راهکارها:

• پیاده‌سازی صف‌های با اولویت
• مسیریابی مبتنی بر موضوع برای رویدادهای مهم

مثال در RabbitMQ:

var args = new Dictionary<string, object> { { &quotx-max-priority&quot, 10 } };
channel.QueueDeclare(&quotpriority-queue&quot, true, false, false, args);

۴. صف مرده (Dead Letter Queue - DLQ)

مشکل: از دست رفتن پیام‌های پردازش نشده

راهکارها:

• انتقال پیام‌های ناموفق به صف جداگانه
• پیاده‌سازی سیاست‌های تکرار پردازش

مثال در RabbitMQ:

var args = new Dictionary<string, object>
{
    { &quotx-dead-letter-exchange&quot, &quotdlx-exchange&quot }
};
channel.QueueDeclare(&quotorder-queue&quot, true, false, false, args);

۵. پردازش دسته‌ای (Batch Processing)

مشکل: ناکارآمدی در پردازش تک‌تک پیام‌ها

راهکارها:

• پردازش گروهی پیام‌ها
• استفاده از Kafka Streams

مثال در Kafka:

public void ProcessBatch(IEnumerable<ConsumeResult<Ignore, string>> messages)
{
    foreach (var message in messages)
    {
        Console.WriteLine($&quotProcessing batch message: {message.Value}&quot);
    }
}

۶. پردازش جریانی (Stream Processing)

مشکل: چالش در پردازش بلادرنگ حجم بالای رویدادها

راهکارها:

• استفاده از Apache Flink یا Kafka Streams
• تکنیک‌های پنجره‌بندی زمانی

مثال در Kafka Streams:

var builder = new StreamsBuilder();
var stream = builder.Stream<string, string>(&quotpayment-stream&quot);

stream.GroupByKey()
    .WindowedBy(TimeWindows.Of(TimeSpan.FromSeconds(5)))
    .Reduce((value1, value2) => value1 + value2);