Какие плюсы и минусы у event-driven архитектуры?

Event-Driven архитектура: плюсы и минусы

Event-driven архитектура — это парадигма проектирования, где компоненты системы взаимодействуют друг с другом через события, а не прямые вызовы функций.

Плюсы Event-Driven архитектуры

1. Слабая связанность (Loose Coupling)

# ❌ Плотная связанность
class OrderProcessor:
    def __init__(self, email_service, inventory_service):
        self.email_service = email_service
        self.inventory_service = inventory_service
    
    def process_order(self, order):
        # Процессор знает о всех деталях
        self.inventory_service.reserve_items(order.items)
        self.email_service.send_confirmation(order.customer_email)
        # Если добавим новый сервис - нужно менять этот класс

# ✅ Event-driven - слабая связанность
class EventBus:
    def __init__(self):
        self.subscribers = {}
    
    def subscribe(self, event_type, handler):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)
    
    def publish(self, event):
        if event.__class__.__name__ in self.subscribers:
            for handler in self.subscribers[event.__class__.__name__]:
                handler(event)

class OrderProcessed:
    def __init__(self, order_id):
        self.order_id = order_id

class OrderProcessor:
    def __init__(self, event_bus):
        self.event_bus = event_bus
    
    def process_order(self, order):
        # Просто публикуем событие
        self.event_bus.publish(OrderProcessed(order.id))
        # Кто слушает - дело их

# Другие сервисы подписываются
event_bus = EventBus()

def send_email(event):
    print(f"Email sent for order {event.order_id}")

def reserve_inventory(event):
    print(f"Inventory reserved for order {event.order_id}")

event_bus.subscribe("OrderProcessed", send_email)
event_bus.subscribe("OrderProcessed", reserve_inventory)

# Добавить новый обработчик просто - не нужно менять processor
def log_to_analytics(event):
    print(f"Analytics: order {event.order_id}")

event_bus.subscribe("OrderProcessed", log_to_analytics)

Преимущество: Новые обработчики добавляются без изменения основного кода.

2. Асинхронная обработка и масштабируемость

import asyncio
from typing import Callable, Any

class AsyncEventBus:
    def __init__(self):
        self.handlers: dict[str, list[Callable]] = {}
    
    def subscribe(self, event_type: str, handler: Callable):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    async def publish(self, event: Any):
        event_type = event.__class__.__name__
        if event_type in self.handlers:
            # Асинхронная обработка нескольких обработчиков
            tasks = [
                self._call_handler(handler, event)
                for handler in self.handlers[event_type]
            ]
            await asyncio.gather(*tasks)
    
    async def _call_handler(self, handler: Callable, event: Any):
        if asyncio.iscoroutinefunction(handler):
            await handler(event)
        else:
            handler(event)

# Обработчики могут быть async
async def send_notification(event):
    await asyncio.sleep(1)  # Имитация I/O
    print(f"Notification sent for {event.order_id}")

async def update_analytics(event):
    await asyncio.sleep(0.5)
    print(f"Analytics updated for {event.order_id}")

# Оба выполнятся параллельно, а не последовательно
bus = AsyncEventBus()
bus.subscribe("OrderProcessed", send_notification)
bus.subscribe("OrderProcessed", update_analytics)

class OrderEvent:
    def __init__(self, order_id):
        self.order_id = order_id

async def main():
    await bus.publish(OrderEvent("order-123"))
    # Обе функции выполнятся одновременно, всего ~1 сек

asyncio.run(main())

3. Реактивность и гибкость

# Одно событие может иметь множество реакций
class PaymentProcessed:
    def __init__(self, order_id, amount):
        self.order_id = order_id
        self.amount = amount

# На одно событие много реакций
event_bus = EventBus()

event_bus.subscribe("PaymentProcessed", lambda e: print(f"Update inventory"))
event_bus.subscribe("PaymentProcessed", lambda e: print(f"Send receipt"))
event_bus.subscribe("PaymentProcessed", lambda e: print(f"Update analytics"))
event_bus.subscribe("PaymentProcessed", lambda e: print(f"Schedule shipping"))

event_bus.publish(PaymentProcessed("order-123", 99.99))
# Все четыре действия выполнятся

4. Тестируемость

class OrderService:
    def __init__(self, event_bus):
        self.event_bus = event_bus
    
    def create_order(self, items):
        order = Order(items)
        self.event_bus.publish(OrderCreated(order.id))
        return order

# Тестирование просто - мокируем event_bus
class MockEventBus:
    def __init__(self):
        self.published_events = []
    
    def publish(self, event):
        self.published_events.append(event)

def test_order_creation():
    mock_bus = MockEventBus()
    service = OrderService(mock_bus)
    
    order = service.create_order(["item1", "item2"])
    
    assert len(mock_bus.published_events) == 1
    assert isinstance(mock_bus.published_events[0], OrderCreated)
    assert mock_bus.published_events[0].order_id == order.id

test_order_creation()
print("Test passed!")

Минусы Event-Driven архитектуры

1. Сложность отладки и трассировки потока выполнения

# ❌ Сложно проследить что произойдет
class UserRegistered:
    def __init__(self, user_id):
        self.user_id = user_id

# Где еще будет обработан это событие?
event_bus.publish(UserRegistered(123))

# Нужно искать все подписки:
# - Может отправить email
# - Может создать профиль
# - Может добавить в рассылку
# - Может отправить SMS
# Если что-то упало - непонятно где

# ✅ Лучше иметь log/отслеживание
class TrackedEventBus:
    def publish(self, event):
        event_type = event.__class__.__name__
        print(f"Publishing {event_type}")
        
        if event_type in self.handlers:
            for i, handler in enumerate(self.handlers[event_type]):
                try:
                    print(f"  -> Handler {i}: {handler.__name__}")
                    handler(event)
                    print(f"  <- Handler {i}: OK")
                except Exception as e:
                    print(f"  <- Handler {i}: ERROR {e}")

2. Race conditions и консистентность данных

# ❌ Проблема: из-за async событий, порядок непредсказуем
class PaymentCompleted:
    def __init__(self, order_id):
        self.order_id = order_id

async def send_email(event):
    # Отправляем email ДО того, как инвентарь обновлен?
    await asyncio.sleep(1)
    print(f"Email sent for {event.order_id}")

async def update_inventory(event):
    await asyncio.sleep(0.5)
    print(f"Inventory updated for {event.order_id}")

# Email может уйти ДО обновления инвентаря
# Пользователь может получить письмо, но товар еще не зарезервирован

# ✅ Решение: явный порядок выполнения
async def send_email_after_inventory(event):
    # Оба события должны завершиться
    # В real-world нужна орхестрация
    pass

3. Сложность с обработкой ошибок

# ❌ Что если обработчик упадет?
def process_event(event):
    handlers = self.handlers.get(event_type, [])
    for handler in handlers:
        handler(event)  # Если упадет - остальные не выполнятся

# ✅ Нужна правильная обработка
async def publish_with_error_handling(self, event):
    event_type = event.__class__.__name__
    
    for handler in self.handlers.get(event_type, []):
        try:
            if asyncio.iscoroutinefunction(handler):
                await handler(event)
            else:
                handler(event)
        except Exception as e:
            # Логируем, но продолжаем обрабатывать остальные
            logging.error(f"Handler {handler.__name__} failed: {e}")
            # Может быть, отправляем в dead letter queue
            await self.handle_failed_event(event, handler, e)

4. Производительность и overhead

# ❌ Каждое событие имеет overhead
for i in range(1_000_000):
    event = Event(i)
    # Поиск подписчиков
    # Сериализация события
    # Очередь сообщений (если async)
    # Десериализация
    # Вызов функции
    event_bus.publish(event)  # Может быть медленнее чем прямой вызов

# ✅ Прямой вызов - быстрее
for i in range(1_000_000):
    function(i)  # Намного быстрее

5. Память и масштабирование

# ❌ При большом количестве подписчиков
class Subject:
    def __init__(self):
        self.observers = []  # Может содержать тысячи обработчиков
    
    def attach(self, observer):
        self.observers.append(observer)
    
    def notify(self):
        for observer in self.observers:
            observer.update()  # Все в памяти одновременно

# ✅ Распределенная обработка с message queues
# RabbitMQ, Kafka и т.д.

Таблица сравнения

Когда использовать Event-Driven

• ✅ Микросервисы - события связывают сервисы
• ✅ Real-time системы - Telegram ботов, веб-сокеты
• ✅ Комплексные бизнес-процессы - заказы, платежи
• ❌ Простые CRUD приложения - может быть overengineering
• ❌ Максимальная производительность - прямые вызовы быстрее

Лучшие практики

• Документируй события - какие события существуют
• Используй DLQ (Dead Letter Queue) для ошибок
• Версионируй события - могут измениться
• Логируй отправку - для отладки
• Тестируй обработчики изолировано
• Монитори порядок выполнения в распределенных системах