Блог AST-SoftPro

Миграция монолита к микросервисам: пошаговый план

05.07.2026 21 мин чтения

Миграция монолита к микросервисам: пошаговый план

Миграция монолитного приложения к микросервисной архитектуре — один из самых сложных процессов в разработке. Ошибки на этом пути приводят к неработающим системам, потере данных и месяцам простоя. В этой статье разберём пошаговый план миграции с примерами на Python.

Почему монолит перестаёт работать

Монолит — хорошая архитектура для старта. Один репозиторий, один процесс, простая отладка. Но по мере роста возникают проблемы:

Независимый деплой невозможен. Изменение одной строки требует пересборки и перезапуска всего приложения.
Масштабирование. Нельзя масштабировать только нагруженный модуль — масштабируется всё приложение целиком.
Технический долг. Старый код тянет за собой устаревшие зависимости, которые нельзя обновить без риска сломать другие модули.
Командная координация. Несколько команд работают в одном коде — конфликты слияний, блокировки, очереди на код-ревью.
Время сборки. Собрать и протестировать монолит на 2+ млн строк кода может занимать часы.

Strangler-fig паттерн

Strangler-fig — паттерн миграции, при котором новая система постепенно заменяет старую. Как фикус-душитель: сначала растёт вокруг старого дерева, потом старое дерево больше не нужно.

Принцип работы:

На фронтенд монолита ставится маршрутизатор (API gateway)
Новые функции реализуются в микросервисах
Существующие функции постепенно переносятся
Когда весь функционал перенесён — монолит отключается

┌──────────────────────────────────────────────┐
│              API Gateway                     │
│                                              │
│  /api/users ──▶  Users Service (новый)      │
│  /api/orders ──▶  Orders Service (новый)    │
│  /api/* ──────▶  Monolith (пока старый)     │
└──────────────────────────────────────────────┘

Шаг 1: Аудит монолита

Перед миграцией нужно понять, что у вас есть. Без аудита миграция — это слепой полёт.

Что исследовать:

Зависимости между модулями. Какие модули вызывают друг друга?
Горячие точки. Какие эндпоинты получают больше всего трафика?
Сложность. Какие модули сложнее всего тестировать и поддерживать?
Владелец. Какая команда отвечает за каждый модуль?

# Скрипт для анализа зависимостей в Django-приложении
import ast
import os

def analyze_imports(project_dir):
    """Анализ импортов между приложениями Django"""
    dependencies = {}

    for root, dirs, files in os.walk(project_dir):
        # Пропускаем виртуальное окружение и миграции
        if 'venv' in root or 'migrations' in root or '__pycache__' in root:
            continue

        for file in files:
            if not file.endswith('.py'):
                continue

            filepath = os.path.join(root, file)
            with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
                try:
                    tree = ast.parse(f.read())
                except SyntaxError:
                    continue

            app_name = os.path.basename(os.path.dirname(filepath))
            if app_name not in dependencies:
                dependencies[app_name] = set()

            for node in ast.walk(tree):
                if isinstance(node, ast.Import):
                    for alias in node.names:
                        parts = alias.name.split('.')
                        if len(parts) > 1:
                            parent_app = parts[0]
                            if parent_app != app_name:
                                dependencies[app_name].add(parent_app)
                elif isinstance(node, ast.ImportFrom):
                    if node.module:
                        parts = node.module.split('.')
                        if len(parts) > 1:
                            parent_app = parts[0]
                            if parent_app != app_name:
                                dependencies[app_name].add(parent_app)

    return dependencies

# Использование
deps = analyze_imports('/path/to/project')
for app, imports in sorted(deps.items()):
    print(f"{app}: {sorted(imports)}")

Шаг 2: Определение bounded contexts

Bounded context — граница доменной области, внутри которой термины имеют единое значение. Это фундамент для разделения монолита.

Пример для интернет-магазина:

Bounded Context	Описание	Сущности
Пользователи	Регистрация, авторизация, профили	User, Profile, Address
Каталог	Товары, категории, поиск	Product, Category, Tag
Заказы	Создание, обработка, статусы	Order, OrderItem, Status
Платежи	Оплата, возвраты, транзакции	Payment, Transaction, Refund
Доставка	Курьеры, отслеживание	Shipment, Tracking, Courier

Инструменты для определения:

Event Storming — воркшоп с командой, где на стикерах раскладываются события домена
Domain-Driven Design — методология для проектирования сложных систем
Anti-Corruption Layer — слой, который защищает новый сервис от модели монолита

Шаг 3: Выбор первого сервиса для выноса

Не пытайтесь перенести всё сразу. Выберите один сервис — тот, который даст максимальную отдачу при минимальном риске.

Критерии выбора:

Низкая связанность. Меньше зависимостей от других модулей — проще вынести.
Чёткая граница. Легко определить входные и выходные данные.
Высокая ценность. Бизнес видит результат быстро.
Ограниченная база данных. Один сервис — одна база данных.

Пример приоритизации:

Сервис	Связанность	Граница	Ценность	Приоритет
Пользователи	Низкая	Чёткая	Средняя	1 (первый)
Каталог	Средняя	Чёткая	Высокая	2
Заказы	Высокая	Чёткая	Высокая	3
Платежи	Низкая	Чёткая	Высокая	4
Доставка	Средняя	Размытая	Средняя	5

Шаг 4: Создание первого микросервиса

Начнём с сервиса пользователей — он имеет низкую связанность и чёткую границу.

# users_service/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

app = FastAPI(title="Users Service")

engine = create_engine("postgresql://localhost/users_db")
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)
    created_at = Column(String)

class UserCreate(BaseModel):
    email: str
    full_name: str

class UserResponse(BaseModel):
    id: int
    email: str
    full_name: str

Base.metadata.create_all(engine)

@app.post("/api/users", response_model=UserResponse)
def create_user(user: UserCreate):
    db = SessionLocal()
    try:
        db_user = User(email=user.email, full_name=user.full_name)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user
    finally:
        db.close()

@app.get("/api/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.id == user_id).first()
        if not user:
            raise HTTPException(404, "User not found")
        return user
    finally:
        db.close()

Шаг 5: Dual-write для миграции данных

Dual-write — паттерн, при котором данные записываются и в монолит, и в новый сервис одновременно. Это обеспечивает консистентность во время миграции.

# Middleware для dual-write
import httpx
import logging

logger = logging.getLogger(__name__)

class DualWriteMiddleware:
    """Записывает данные и в монолит, и в новый сервис"""

    def __init__(self, service_url: str):
        self.service_url = service_url
        self.client = httpx.AsyncClient(timeout=5.0)

    async def write_to_service(self, method: str, path: str, data: dict):
        """Асинхронная запись в новый сервис без блокировки"""
        try:
            url = f"{self.service_url}{path}"
            response = await self.client.request(method, url, json=data)
            response.raise_for_status()
        except Exception as e:
            # Логируем, но не блокируем основной поток
            logger.error(f"Dual-write failed for {method} {path}: {e}")

# Использование в монолите
user_service = DualWriteMiddleware("http://users-service:8000")

@app.post("/api/users")
def create_user(request):
    # 1. Запись в монолит (основная)
    user = User.objects.create(
        email=request.email,
        full_name=request.full_name,
    )

    # 2. Запись в новый сервис (асинхронно)
    asyncio.create_task(
        user_service.write_to_service(
            "POST", "/api/users",
            {"email": user.email, "full_name": user.full_name},
        )
    )

    return {"id": user.id, "email": user.email}

Шаг 6: Feature toggles

Feature toggles позволяют включать и отключать функциональность без деплоя кода. Это критически важно для плавной миграции.

# feature_toggles.py
import os

FEATURES = {
    "users_service_enabled": os.getenv("USERS_SERVICE_ENABLED", "false").lower() == "true",
    "catalog_service_enabled": os.getenv("CATALOG_SERVICE_ENABLED", "false").lower() == "true",
    "orders_service_enabled": os.getenv("ORDERS_SERVICE_ENABLED", "false").lower() == "true",
}

def toggle(feature_name: str) -> bool:
    return FEATURES.get(feature_name, False)

# Использование в маршрутизаторе
from fastapi import APIRouter

router = APIRouter()

@router.get("/api/users/{user_id}")
async def get_user(user_id: int):
    if toggle("users_service_enabled"):
        # Новый сервис
        async with httpx.AsyncClient() as client:
            response = await client.get(f"http://users-service:8000/api/users/{user_id}")
            return response.json()
    else:
        # Монолит
        user = User.objects.get(id=user_id)
        return {"id": user.id, "email": user.email, "full_name": user.full_name}

Шаг 7: Миграция базы данных

Каждый микросервис должен иметь свою собственную базу данных. Это принцип отдельной базы для каждого сервиса.

Стратегии миграции данных:

CDC — отслеживание изменений в базе данных монолита и репликация в новый сервис
Dual-write — запись в обе базы одновременно (описано выше)
Big Bang — разовый перенос данных (только для чтения)

Пример CDC с Debezium:

# Потребитель CDC-событий
from kafka import KafkaConsumer
import json

class CDCConsumer:
    """Потребляет события изменений из Kafka и применяет в новый сервис"""

    def __init__(self, bootstrap_servers: str, topic: str):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            auto_offset_reset='earliest',
            group_id='migration-group',
        )

    def process(self):
        for message in self.consumer:
            event = message.value
            operation = event.get('op')  # 'c' = create, 'u' = update, 'd' = delete

            if operation == 'c':
                self.handle_create(event['after'])
            elif operation == 'u':
                self.handle_update(event['after'])
            elif operation == 'd':
                self.handle_delete(event['before']['id'])

    def handle_create(self, data):
        # Вставка в новый сервис
        pass

    def handle_update(self, data):
        # Обновление в новом сервисе
        pass

    def handle_delete(self, user_id):
        # Удаление из нового сервиса
        pass

Шаг 8: Настройка коммуникации

Микросервисы общаются друг с другом через API. Есть два подхода:

Синхронная (REST/gRPC) — запрос-ответ, подходит для операций, где нужен немедленный результат
Асинхронная (Kafka/RabbitMQ) — события, подходит для операций, где результат не нужен сразу

# Пример асинхронной коммуникации через события
import pika
import json

class EventBus:
    """Шина событий на базе RabbitMQ"""

    def __init__(self, connection_string: str):
        self.connection = pika.BlockingConnection(
            pika.URLParameters(connection_string)
        )
        self.channel = self.connection.channel()
        self.channel.exchange_declare(
            exchange='events',
            exchange_type='topic',
            durable=True,
        )

    def publish(self, event_type: str, data: dict):
        message = json.dumps({
            "event_type": event_type,
            "timestamp": __import__('datetime').datetime.utcnow().isoformat(),
            "data": data,
        })
        self.channel.basic_publish(
            exchange='events',
            routing_key=event_type,
            body=message,
            properties=pika.BasicProperties(delivery_mode=2),  # persistent
        )

    def subscribe(self, event_type: str, handler):
        queue = self.channel.queue_declare(queue=f"handler_{event_type}").method.queue
        self.channel.queue_bind(
            exchange='events',
            queue=queue,
            routing_key=event_type,
        )
        self.channel.basic_consume(queue=queue, on_message_callback=handler)
        self.channel.start_consuming()

# Использование
bus = EventBus("amqp://localhost")

# Публикация события
bus.publish("order.created", {"order_id": 123, "total": 999.99})

# Подписка на событие
def on_order_created(ch, method, properties, body):
    event = json.loads(body)
    # Отправить email, обновить аналитику и т.д.
    pass

bus.subscribe("order.created", on_order_created)

Шаг 9: Наблюдаемость

В микросервисной архитектуре наблюдаемость критически важна. Без неё отладка превращается в кошмар.

Distributed tracing — Jaeger или Zipkin для отслеживания запросов через сервисы
Централизованное логирование — ELK Stack или Loki для сбора логов
Метрики — Prometheus + Grafana для мониторинга
Health checks — каждый сервис должен отдавать /health

# Health check для микросервиса
from fastapi import FastAPI
from sqlalchemy import text

app = FastAPI()

@app.get("/health")
def health_check():
    health = {"status": "healthy", "services": {}}

    # Проверка базы данных
    try:
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        health["services"]["database"] = "healthy"
    except Exception:
        health["services"]["database"] = "unhealthy"
        health["status"] = "degraded"

    # Проверка Redis
    try:
        redis.ping()
        health["services"]["redis"] = "healthy"
    except Exception:
        health["services"]["redis"] = "unhealthy"
        health["status"] = "degraded"

    status_code = 200 if health["status"] == "healthy" else 503
    return JSONResponse(content=health, status_code=status_code)

Чек-лист миграции

[ ] Аудит зависимостей завершён
[ ] Bounded contexts определены
[ ] Первый сервис выбран (низкая связанность)
[ ] API Gateway настроен
[ ] Dual-write реализован
[ ] Feature toggles работают
[ ] База данных мигрирована
[ ] Коммуникация между сервисами настроена
[ ] Наблюдаемость (tracing, logging, metrics) работает
[ ] Health checks настроены
[ ] Load testing проведён
[ ] План отката написан

Частые ошибки

Миграция ради миграции. Если монолит работает — не трогайте его. Миграция стоит денег и времени.
Распределённый монолит. Вынесли сервисы, но они всё ещё тесно связаны и деплоятся вместе. Это не микросервисы — это монолит, размазанный по серверам.
Совместная база данных. Несколько сервисов пишут в одну базу — это антипаттерн. Каждый сервис — своя база.
Отсутствие плана отката. Если что-то пошло не так, нужно уметь быстро вернуться к монолиту.
Слишком мелкие сервисы. 50 микросервисов для команды из 10 человек — это не архитектура, это административный кошмар.

Заключение

Миграция монолита к микросервисам — это марафон, а не спринт. Strangler-fig паттерн позволяет мигрировать постепенно, без простоев. Dual-write и feature toggles обеспечивают плавный переход. А наблюдаемость — ваш главный инструмент в новой архитектуре.

Главное правило: мигрируйте только тогда, когда монолит действительно мешает. И мигрируйте по одному сервису за раз.

Ключевые моменты:

Strangler-fig — основной паттерн миграции: постепенно заменяем монолит через API Gateway.
Bounded contexts определяют границы микросервисов — каждый сервис отвечает за свою доменную область.
Dual-write обеспечивает консистентность данных во время миграции.
Feature toggles позволяют плавно переключать трафик без деплоя.
Каждый микросервис — своя база данных. Совместная база — антипаттерн.

Другие записи блога

Все статьи

К списку статей

Миграция монолита к микросервисам: пошаговый план