Что такое API-композиция в микросервисной архитектуре?

API-композиция в микросервисной архитектуре

API-композиция — это паттерн проектирования, который позволяет объединять данные и функциональность из нескольких микросервисов в единый ответ для клиента. Вместо того чтобы клиент сам вызывал каждый микросервис отдельно, существует промежуточный слой (API Gateway или отдельный сервис), который координирует эти вызовы и собирает результаты.

Проблема, которую решает API-композиция

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

• Множественные запросы — клиент должен делать несколько HTTP-запросов
• Задержки сети — каждый запрос добавляет latency
• Сложная логика на клиенте — клиент управляет координацией
• Проблемы с отказоустойчивостью — если один сервис недоступен, вся операция может провалиться

Подходы к API-композиции

1. Оркестрация (API Gateway)

Центральный компонент (API Gateway) явно координирует запросы к микросервисам:

# Пример с FastAPI как API Gateway
from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/users/{user_id}/profile")
async def get_user_profile(user_id: str):
    async with httpx.AsyncClient() as client:
        # Запрос к сервису пользователей
        user_resp = await client.get(f"http://user-service/users/{user_id}")
        user_data = user_resp.json()
        
        # Запрос к сервису заказов
        orders_resp = await client.get(f"http://order-service/users/{user_id}/orders")
        orders_data = orders_resp.json()
        
        # Запрос к сервису платежей
        payments_resp = await client.get(f"http://payment-service/users/{user_id}/payments")
        payments_data = payments_resp.json()
        
        # Композиция результатов
        return {
            "user": user_data,
            "orders": orders_data,
            "payments": payments_data
        }

2. Хореография (Event-driven)

Микросервисы реагируют на события и асинхронно отправляют данные. Клиент получает данные через WebSocket или polling:

# Микросервис заказов публикует событие
from pydantic import BaseModel
import aio_pika

class OrderCreatedEvent(BaseModel):
    order_id: str
    user_id: str
    amount: float

async def publish_order_event(event: OrderCreatedEvent):
    connection = await aio_pika.connect_robust("amqp://guest:guest@rabbitmq/")
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange(
            "orders_exchange", aio_pika.ExchangeType.TOPIC
        )
        message = aio_pika.Message(body=event.json().encode())
        await exchange.publish(message, routing_key="order.created")

Паттерны для обработки параллельных запросов

import asyncio
from typing import List

async def fetch_from_services(user_id: str):
    """Параллельное выполнение запросов с asyncio"""
    async with httpx.AsyncClient() as client:
        # Создаём корутины для всех запросов
        user_task = client.get(f"http://user-service/users/{user_id}")
        orders_task = client.get(f"http://order-service/users/{user_id}/orders")
        payments_task = client.get(f"http://payment-service/users/{user_id}/payments")
        
        # Выполняем их параллельно
        user_resp, orders_resp, payments_resp = await asyncio.gather(
            user_task, orders_task, payments_task
        )
        
        return {
            "user": user_resp.json(),
            "orders": orders_resp.json(),
            "payments": payments_resp.json()
        }

Обработка ошибок и отказоустойчивость

from tenacity import retry, stop_after_attempt, wait_exponential
from circuitbreaker import circuit

class ServiceComposer:
    @circuit(failure_threshold=5, recovery_timeout=60)
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def fetch_from_service(self, url: str):
        """Запрос с retry и circuit breaker"""
        async with httpx.AsyncClient() as client:
            response = await client.get(url, timeout=5.0)
            response.raise_for_status()
            return response.json()
    
    async def get_user_profile(self, user_id: str):
        """Обработка ошибок отдельных сервисов"""
        try:
            user = await self.fetch_from_service(f"http://user-service/users/{user_id}")
        except Exception as e:
            user = {"error": str(e)}
        
        try:
            orders = await self.fetch_from_service(f"http://order-service/users/{user_id}/orders")
        except Exception as e:
            orders = []
        
        return {"user": user, "orders": orders}

Преимущества и недостатки

Преимущества:

• Снижение нагрузки на клиент
• Единая точка контроля и логирования
• Возможность кэширования результатов
• Улучшение производительности благодаря параллелизму

Недостатки:

• Дополнительная сложность в архитектуре
• Риск создания нового monolith'а в виде API Gateway
• Сложность отладки и трассировки
• Проблема N+1 запросов

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

1. Используй асинхронность — async/await для параллельных запросов
2. Кэширование — сокращает количество запросов к микросервисам
3. Таймауты и retry — защищает от зависания
4. Circuit breaker — предотвращает каскадные сбои
5. Мониторинг и логирование — трассировка запросов между сервисами (distributed tracing)
6. Ограничение глубины композиции — избегай глубоких цепочек зависимостей

API-композиция — важный паттерн в микросервисной архитектуре, который при правильном применении значительно улучшает пользовательский опыт и масштабируемость системы.