Какие есть способы взаимодействия компонентов в виде сервисов?

Способы взаимодействия сервисов в микросервисной архитектуре

Взаимодействие между сервисами — критичный аспект микросервисной архитектуры. Есть несколько основных подходов:

1. Синхронное взаимодействие (Request-Response)

1.1 REST API (HTTP)

Самый простой и распространённый способ.

# Service A вызывает Service B
import requests
from typing import Dict, Any

class UserService:
    def __init__(self, base_url: str = "http://user-service:8001"):
        self.base_url = base_url
    
    def get_user(self, user_id: int) -> Dict[str, Any]:
        try:
            response = requests.get(
                f"{self.base_url}/api/v1/users/{user_id}",
                timeout=5
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            raise ServiceError(f"Failed to get user: {e}")

# OrderService использует UserService
class OrderService:
    def __init__(self, user_service: UserService):
        self.user_service = user_service
    
    def create_order(self, user_id: int, items: list) -> Dict:
        # Проверяем, что пользователь существует
        user = self.user_service.get_user(user_id)
        
        return {
            "user_id": user_id,
            "items": items,
            "total": sum(item["price"] for item in items)
        }

Плюсы:

• Простота реализации
• Инструменты для отладки
• Стандартный протокол (HTTP)

Минусы:

• Tight coupling (сильная связанность)
• Проблемы с reliability (если B отвалится, А зависает)
• N+1 problem (множество запросов)

1.2 gRPC (protobuf)

Высокопроизводительный протокол RPC на базе HTTP/2.

// user_service.proto
syntax = "proto3";

service UserService {
    rpc GetUser (GetUserRequest) returns (User);
    rpc ListUsers (ListUsersRequest) returns (UserList);
}

message GetUserRequest {
    int32 user_id = 1;
}

message User {
    int32 id = 1;
    string name = 2;
    string email = 3;
}

message UserList {
    repeated User users = 1;
}

# Python сервис
from grpc import aio
import user_service_pb2
import user_service_pb2_grpc

class UserService(user_service_pb2_grpc.UserServiceServicer):
    async def GetUser(self, request, context):
        user = await self.db.get_user(request.user_id)
        return user_service_pb2.User(
            id=user['id'],
            name=user['name'],
            email=user['email']
        )

# Клиент
async def call_user_service():
    async with aio.secure_channel('localhost:50051', ...) as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        user = await stub.GetUser(
            user_service_pb2.GetUserRequest(user_id=123)
        )
        return user

Плюсы:

• Высокая производительность (бинарный протокол)
• Streaming поддержка -強типизация

Минусы:

• Сложнее чем REST
• Менее стандартное для веб-приложений

2. Асинхронное взаимодействие (Event-Driven)

2.1 Message Queue (RabbitMQ, Kafka)

Сервисы обмениваются сообщениями через очередь.

# Service A публикует событие
import pika
import json

def publish_order_created(order_id: int, user_id: int):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Декларируем exchange
    channel.exchange_declare(
        exchange='orders',
        exchange_type='topic',
        durable=True
    )
    
    # Публикуем событие
    event = {
        "event_type": "order.created",
        "order_id": order_id,
        "user_id": user_id,
        "timestamp": "2024-03-22T10:00:00Z"
    }
    
    channel.basic_publish(
        exchange='orders',
        routing_key='order.created',
        body=json.dumps(event),
        properties=pika.BasicProperties(
            delivery_mode=2,  # Persistent
            content_type='application/json'
        )
    )
    
    connection.close()

# Service B слушает событие
def start_order_listener():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # Декларируем queue
    channel.exchange_declare(exchange='orders', exchange_type='topic', durable=True)
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    
    # Биндим queue к exchange
    channel.queue_bind(
        exchange='orders',
        queue=queue_name,
        routing_key='order.*'
    )
    
    def callback(ch, method, properties, body):
        event = json.loads(body)
        print(f"Received event: {event}")
        
        # Обрабатываем событие
        handle_order_event(event)
        
        ch.basic_ack(delivery_tag=method.delivery_tag)
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback
    )
    
    channel.start_consuming()

def handle_order_event(event):
    if event["event_type"] == "order.created":
        # Отправляем email, обновляем инвентарь и т.д.
        send_confirmation_email(event["user_id"])
        update_inventory(event["order_id"])

Плюсы:

• Loose coupling (слабая связанность)
• Масштабируемость (можно добавить слушателей)
• Reliability (можно переиспользовать сообщения)

Минусы:

• Сложнее отладка
• Возможны дублирующиеся сообщения
• Задержка в доставке

2.2 Event Sourcing + CQRS

Сохранение всех изменений как событий.

from dataclasses import dataclass
from datetime import datetime
from typing import List
import json

@dataclass
class Event:
    event_type: str
    aggregate_id: str
    data: dict
    timestamp: datetime

class EventStore:
    def __init__(self):
        self.events: List[Event] = []
    
    def append(self, event: Event):
        self.events.append(event)
    
    def get_events(self, aggregate_id: str) -> List[Event]:
        return [e for e in self.events if e.aggregate_id == aggregate_id]

# Сохраняем каждое изменение как событие
class OrderAggregate:
    def __init__(self, order_id: str, event_store: EventStore):
        self.order_id = order_id
        self.event_store = event_store
        self.state = self._rebuild_state()
    
    def _rebuild_state(self) -> dict:
        """Восстанавливаем состояние из всех событий"""
        state = {
            "id": self.order_id,
            "status": "initial",
            "items": [],
            "total": 0
        }
        
        for event in self.event_store.get_events(self.order_id):
            if event.event_type == "order.created":
                state["status"] = "created"
                state["items"] = event.data["items"]
            elif event.event_type == "order.payment_confirmed":
                state["status"] = "paid"
            elif event.event_type == "order.shipped":
                state["status"] = "shipped"
        
        return state
    
    def create_order(self, items: list, total: float):
        event = Event(
            event_type="order.created",
            aggregate_id=self.order_id,
            data={"items": items, "total": total},
            timestamp=datetime.now()
        )
        self.event_store.append(event)
        self.state = self._rebuild_state()
    
    def confirm_payment(self):
        event = Event(
            event_type="order.payment_confirmed",
            aggregate_id=self.order_id,
            data={},
            timestamp=datetime.now()
        )
        self.event_store.append(event)
        self.state = self._rebuild_state()

# Использование
event_store = EventStore()
order = OrderAggregate("order-123", event_store)
order.create_order([{"sku": "A1", "qty": 2}], 99.99)
order.confirm_payment()

# История всех событий доступна для воспроизведения
print(order.state)  # Восстановлено из событий

Плюсы:

• Full audit trail (полная история изменений)
• Временная машина (можно вернуться в прошлое)
• Natural event-driven

Минусы:

• Complexity (сложная для понимания)
• Storage overhead (много событий)
• Consistency challenges

3. Гибридные подходы

3.1 Saga pattern

Координирует долгоживущие транзакции между сервисами.

# Orchestration Saga (Choreo pattern)
class OrderSaga:
    def __init__(self, event_bus, services):
        self.event_bus = event_bus
        self.payment_service = services.payment
        self.inventory_service = services.inventory
        self.shipping_service = services.shipping
    
    def execute_order(self, order: dict):
        try:
            # Шаг 1: Зарезервировать товар
            self.inventory_service.reserve(order["items"])
            self.event_bus.publish("order.inventory_reserved", order)
            
            # Шаг 2: Обработать платёж
            self.payment_service.charge(order["user_id"], order["total"])
            self.event_bus.publish("order.payment_processed", order)
            
            # Шаг 3: Отправить заказ
            self.shipping_service.ship(order["id"])
            self.event_bus.publish("order.shipped", order)
            
        except Exception as e:
            # Компенсирующие операции при ошибке
            self.inventory_service.release(order["items"])
            self.event_bus.publish("order.compensated", order)
            raise

Плюсы:

• Контролируемые транзакции между сервисами
• Компенсирующие операции при ошибках

Минусы:

• Сложная логика
• Сложно отладить

4. Сравнительная таблица

Рекомендации

1. Начни с REST — просто для простого взаимодействия
2. Переходи на Message Queue — когда нужна масштабируемость
3. Используй Saga — для долгоживущих процессов
4. Event Sourcing — когда нужна полная история