Event-Driven Architecture

╔═══════════════════════════════════════════════════════════════════════════════╗
║                    EVENT-DRIVEN MICROSERVICES ARCHITECTURE                     ║
╚═══════════════════════════════════════════════════════════════════════════════╝

                        ┌─────────────┐
                        │   Client    │
                        │  (Web/App)  │
                        └──────┬──────┘
                               │
                               ▼
                        ┌─────────────┐
                        │ API Gateway │
                        └──────┬──────┘
                               │
  ┌────────────────────────────┼────────────────────────────┐
  │                            │                            │
  ▼                            ▼                            ▼
┌─────────────┐          ┌─────────────┐          ┌─────────────┐
│   Order     │          │   User      │          │  Product    │
│  Service    │          │  Service    │          │  Service    │
└──────┬──────┘          └──────┬──────┘          └──────┬──────┘
  │                            │                            │
  │ publish                    │ publish                    │ publish
  ▼                            ▼                            ▼
═══════════════════════════════════════════════════════════════════════════════
║                                                                               ║
║                          📡  EVENT BROKER  📡                                 ║
║                                                                               ║
║    ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐             ║
║    │  orders.created │  │  users.created  │  │ products.updated│             ║
║    │     Topic       │  │     Topic       │  │     Topic       │             ║
║    └─────────────────┘  └─────────────────┘  └─────────────────┘             ║
║                                                                               ║
═══════════════════════════════════════════════════════════════════════════════
  │                            │                            │
  │ subscribe                  │ subscribe                  │ subscribe
  ▼                            ▼                            ▼
┌─────────────┐          ┌─────────────┐          ┌─────────────┐
│  Payment    │          │Notification │          │  Analytics  │
│  Service    │          │  Service    │          │  Service    │
└─────────────┘          └─────────────┘          └─────────────┘
  │                            │                            │
  ▼                            ▼                            ▼
┌─────────────┐          ┌─────────────┐          ┌─────────────┐
│  Payment DB │          │ Email/Push  │          │   ClickHouse│
└─────────────┘          └─────────────┘          └─────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                              LEGEND                                         │
├─────────────────────────────────────────────────────────────────────────────┤
│  █ Producer  - сервисы, публикующие события                                │
│  █ Consumer  - сервисы, обрабатывающие события                              │
│  █ Broker    - Kafka / RabbitMQ / Redis Streams                            │
│  █ Storage   - базы данных, внешние системы                                │
│  █ Gateway   - точка входа (Kong, Nginx, etc.)                             │
└─────────────────────────────────────────────────────────────────────────────┘

version: '3.8'

services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- "2181:2181"

kafka:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

kafka-ui:
image: provectuslabs/kafka-ui:latest
ports:
- "8080:8080"
environment:
KAFKA_CLUSTERS_0_NAME: local
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

from confluent_kafka import Producer
import json
from uuid import uuid4
from datetime import datetime

class KafkaEventProducer:
"""Producer событий для Kafka"""

def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.producer = Producer({
'bootstrap.servers': bootstrap_servers,
'client.id': 'order-service'
})

def _delivery_report(self, err, msg):
"""Callback при доставке сообщения"""
if err:
print(f'❌ Delivery failed: {err}')
else:
print(f'✅ Delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

def publish(self, topic: str, event_type: str, data: dict, key: str = None):
"""Опубликовать событие в Kafka"""
event = {
"event_id": str(uuid4()),
"event_type": event_type,
"timestamp": datetime.utcnow().isoformat() + "Z",
"source": "order-service",
"data": data
}

# Key определяет партицию (важно для порядка)
message_key = key or event["event_id"]

self.producer.produce(
topic=topic,
key=message_key.encode('utf-8'),
value=json.dumps(event).encode('utf-8'),
callback=self._delivery_report
)

# Flush для гарантии отправки
self.producer.flush()

return event

# ═══════════════════════════════════════════════════════════════
#                         ИСПОЛЬЗОВАНИЕ
# ═══════════════════════════════════════════════════════════════

if __name__ == "__main__":
producer = KafkaEventProducer()

# Публикуем событие создания заказа
event = producer.publish(
topic="orders",
event_type="OrderCreated",
data={
"order_id": "ord_12345",
"user_id": "usr_789",
"items": [
    {"product_id": "p1", "quantity": 2, "price": 29.99}
],
"total": 59.98
},
key="ord_12345"  # Все события заказа в одной партиции
)

print(f"📤 Published: {event['event_type']}")

from confluent_kafka import Consumer, KafkaError
import json

class KafkaEventConsumer:
"""Consumer событий из Kafka"""

def __init__(
self, 
bootstrap_servers: str = "localhost:9092",
group_id: str = "payment-service"
):
self.consumer = Consumer({
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest',  # Читать с начала
'enable.auto.commit': False       # Ручной commit
})
self.handlers = {}

def subscribe(self, event_type: str):
"""Декоратор для подписки на тип события"""
def decorator(handler):
self.handlers[event_type] = handler
return handler
return decorator

def start(self, topics: list):
"""Запустить прослушивание топиков"""
self.consumer.subscribe(topics)
print(f"🎧 Listening to topics: {topics}")

try:
while True:
    msg = self.consumer.poll(timeout=1.0)

    if msg is None:
        continue

    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        print(f"❌ Error: {msg.error()}")
        continue

    # Парсим событие
    event = json.loads(msg.value().decode('utf-8'))
    event_type = event.get("event_type")

    print(f"📥 Received: {event_type} from {msg.topic()}")

    # Вызываем обработчик
    if event_type in self.handlers:
        try:
            self.handlers[event_type](event)
            # Подтверждаем обработку
            self.consumer.commit(asynchronous=False)
        except Exception as e:
            print(f"❌ Handler error: {e}")
            # Здесь можно отправить в DLQ
    else:
        print(f"⚠️ No handler for {event_type}")
        self.consumer.commit(asynchronous=False)

except KeyboardInterrupt:
print("👋 Shutting down...")
finally:
self.consumer.close()

# ═══════════════════════════════════════════════════════════════
#                      PAYMENT SERVICE
# ═══════════════════════════════════════════════════════════════

consumer = KafkaEventConsumer(group_id="payment-service")

@consumer.subscribe("OrderCreated")
def handle_order_created(event: dict):
"""Обработка нового заказа — инициация оплаты"""
data = event.get("data", {})
order_id = data.get("order_id")
total = data.get("total")

print(f"💳 Processing payment for order {order_id}...")
print(f"💳 Amount: ${total}")

# Здесь логика оплаты...

print(f"💳 ✅ Payment successful!")

if __name__ == "__main__":
consumer.start(topics=["orders"])

import logging
import json
from uuid import uuid4
from datetime import datetime
from contextvars import ContextVar

# Context для трейсинга
correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default='')

class StructuredLogger:
"""Структурированный логгер с correlation_id"""

def __init__(self, name: str):
self.logger = logging.getLogger(name)
self.service = name

def _log(self, level: str, message: str, **extra):
log_entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": level,
"service": self.service,
"correlation_id": correlation_id_var.get(),
"message": message,
**extra
}
print(json.dumps(log_entry))

def info(self, message: str, **extra):
self._log("INFO", message, **extra)

def error(self, message: str, **extra):
self._log("ERROR", message, **extra)

def event_received(self, event: dict):
"""Логирование полученного события"""
# Устанавливаем correlation_id из события
corr_id = event.get("correlation_id") or event.get("event_id")
correlation_id_var.set(corr_id)

self._log("INFO", "Event received", 
event_type=event.get("event_type"),
event_id=event.get("event_id")
)

def event_processed(self, event: dict, duration_ms: float):
"""Логирование обработанного события"""
self._log("INFO", "Event processed",
event_type=event.get("event_type"),
event_id=event.get("event_id"),
duration_ms=duration_ms
)

# Использование
logger = StructuredLogger("payment-service")

# При получении события
# logger.event_received(event)
# ... обработка ...
# logger.event_processed(event, duration_ms=45.2)

╔═══════════════════════════════════════════════════════════════════════════╗
║                  EVENT-DRIVEN ARCHITECTURE CHEAT SHEET                     ║
╚═══════════════════════════════════════════════════════════════════════════╝

┌─────────────────────────────────────────────────────────────────────────────┐
│ КОМПОНЕНТЫ                                                                  │
├─────────────────────────────────────────────────────────────────────────────┤
│ Producer      → Создаёт и публикует события                               │
│ Event         → Неизменяемый факт о произошедшем                          │
│ Broker        → Хранит и маршрутизирует события (Kafka, RabbitMQ)         │
│ Consumer      → Подписывается и обрабатывает события                      │
│ Topic/Queue   → Канал для определённого типа событий                      │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│ ПАТТЕРНЫ                                                                    │
├─────────────────────────────────────────────────────────────────────────────┤
│ Pub/Sub           → Один-ко-многим, все подписчики получают копию        │
│ Message Queue     → Point-to-point, сообщение = один consumer             │
│ Event Sourcing    → Состояние = сумма всех событий                        │
│ CQRS              → Разделение записи и чтения                            │
│ Saga              → Распределённая транзакция через события               │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│ СТРУКТУРА СОБЫТИЯ                                                           │
├─────────────────────────────────────────────────────────────────────────────┤
│  {                                                                          │
│    "event_id": "uuid",           // Уникальный ID                        │
│    "event_type": "OrderCreated", // Тип (прошедшее время!)               │
│    "timestamp": "ISO8601",       // Когда произошло                      │
│    "source": "order-service",    // Кто создал                           │
│    "correlation_id": "req_id",  // Для трейсинга                         │
│    "data": { ... }               // Полезная нагрузка                     │
│  }                                                                          │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│ ВЫБОР БРОКЕРА                                                               │
├─────────────────────────────────────────────────────────────────────────────┤
│ Kafka       → High throughput, replay, event sourcing, Big Data          │
│ RabbitMQ    → Сложная маршрутизация, ACK, традиционные очереди           │
│ Redis       → Простота, low latency, уже используете Redis               │
│ AWS SQS/SNS → Serverless, managed, AWS экосистема                       │
│ Google Pub/Sub → GCP экосистема, global, managed                        │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│ ГАРАНТИИ ДОСТАВКИ                                                           │
├─────────────────────────────────────────────────────────────────────────────┤
│ At-most-once   → Fire-and-forget, может потеряться                       │
│ At-least-once  → Гарантия доставки, возможны дубликаты                   │
│ Exactly-once   → Идеально, но сложно (idempotency + deduplication)       │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│ BEST PRACTICES                                                              │
├─────────────────────────────────────────────────────────────────────────────┤
│ ✅ Именуйте события в прошедшем времени (OrderCreated, NOT CreateOrder)   │
│ ✅ Включайте correlation_id для трейсинга                                 │
│ ✅ Делайте consumer'ы идемпотентными                                      │
│ ✅ Версионируйте схемы событий                                            │
│ ✅ Используйте Dead Letter Queue для ошибок                               │
│ ✅ Мониторьте consumer lag                                                │
│ ❌ Не кладите огромные payload'ы в события                                │
│ ❌ Не полагайтесь на порядок событий между партициями                     │
└─────────────────────────────────────────────────────────────────────────────┘