Как передаётся ID от одного сервиса к другому?

Как передаётся ID от одного сервиса к другому

Передача ID между микросервисами — ключевой вопрос в распределённых системах. Разберёмся со всеми подходами и их плюсами/минусами.

Проблема

Когда у тебя несколько независимых сервисов, они должны договориться о том, как идентифицировать объекты. Простой автоинкремент не подходит, потому что может совпасть в разных базах.

Способ 1: UUID (Universally Unique Identifier)

Мост практичный способ для микросервисов:

# Service A: Users Service
from uuid import uuid4
from sqlalchemy import Column, String

class User(Base):
    __tablename__ = 'users'
    id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
    name = Column(String)

# Создаём пользователя
user = User(name='Alice')
session.add(user)
session.commit()
print(user.id)  # a1b2c3d4-e5f6-4g7h-8i9j-k0l1m2n3o4p5

# Service B получает ID в API
response = requests.get(
    'https://users-service/api/users',
    headers={'Authorization': 'Bearer token'}
)
user_data = response.json()
user_id = user_data['id']  # a1b2c3d4-e5f6-4g7h-8i9j-k0l1m2n3o4p5

# Service B может использовать этот ID в своей БД
class Order(Base):
    __tablename__ = 'orders'
    id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
    user_id = Column(String(36))  # Ссылка на User из другого сервиса
    amount = Column(Float)

order = Order(user_id=user_id, amount=99.99)
session.add(order)
session.commit()

Плюсы:

• Генерируется независимо, не нужна синхронизация
• Безопаснее (сложнее предугадать ID)
• Стандарт для микросервисов

Минусы:

• Больше памяти (36 символов vs 8)
• Медленнее для индексирования (чуть)

Способ 2: Nanoid (Shorter UUID alternative)

Как UUID, но короче:

from nanoid import generate

class User(Base):
    __tablename__ = 'users'
    id = Column(String(21), primary_key=True, default=generate)  # 21 символ
    name = Column(String)

user = User(name='Bob')
session.add(user)
session.commit()
print(user.id)  # V1StGXR_Z5j3EqaT

Установка:

pip install python-nanoid

Способ 3: Snowflake ID (для высоконагруженных систем)

Распределённый ID, похож на то, что используют Twitter, Instagram:

from snowflake import SnowflakeGenerator

sg = SnowflakeGenerator(worker_id=1, datacenter_id=1)

class Post(Base):
    __tablename__ = 'posts'
    id = Column(BigInteger, primary_key=True, default=sg.next)
    user_id = Column(BigInteger)
    content = Column(String)

# Использование
post = Post(user_id=123456789, content='Hello')
session.add(post)
session.commit()
print(post.id)  # 7654321098765432

Структура Snowflake ID:

| Timestamp (41 бит) | Datacenter (5 бит) | Worker (5 бит) | Sequence (12 бит) |
| 13 лет            | 32 центра          | 32 воркера     | 4096 ID/мс        |

Способ 4: ID передача через REST API

Основной сценарий в микросервисах:

# Service A: Users API
from fastapi import FastAPI, HTTPException
from uuid import uuid4

app = FastAPI()

users_db = {}

@app.post('/users')
def create_user(name: str):
    user_id = str(uuid4())
    users_db[user_id] = {'id': user_id, 'name': name}
    return {
        'id': user_id,
        'name': name
    }

@app.get('/users/{user_id}')
def get_user(user_id: str):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail='User not found')
    return users_db[user_id]

# Service B: Orders Service
import requests

def create_order(user_id: str, amount: float):
    # Проверяем, что пользователь существует в Service A
    response = requests.get(
        f'https://users-service/users/{user_id}'
    )
    
    if response.status_code != 200:
        raise ValueError(f'User {user_id} not found')
    
    user_data = response.json()
    
    # Создаём заказ со ссылкой на пользователя
    order = {
        'id': str(uuid4()),
        'user_id': user_id,  # Используем ID от Service A
        'amount': amount
    }
    
    return order

Способ 5: ID передача через message queue (async)

Для асинхронной коммуникации:

# Service A: Event Publisher
import json
import pika
from uuid import uuid4

connection = pika.BlockingConnection(
    pika.ConnectionParameters('rabbitmq')
)
channel = connection.channel()
channel.exchange_declare(exchange='events', exchange_type='topic')

@app.post('/users')
def create_user(name: str):
    user_id = str(uuid4())
    
    # Сохраняем в БД
    # ... save to database ...
    
    # Публикуем событие с ID
    event = {
        'event_type': 'user.created',
        'user_id': user_id,
        'name': name,
        'timestamp': '2025-03-22T10:30:00Z'
    }
    
    channel.basic_publish(
        exchange='events',
        routing_key='user.created',
        body=json.dumps(event)
    )
    
    return {'id': user_id}

# Service B: Event Consumer
def on_user_created(ch, method, properties, body):
    event = json.loads(body)
    user_id = event['user_id']  # Используем ID от Service A
    
    # Создаём профиль в Service B
    profile = {
        'id': str(uuid4()),  # Свой ID для профиля
        'user_id': user_id,  # Ссылка на пользователя
        'created_at': event['timestamp']
    }
    
    # Сохраняем профиль
    # ... save to database ...
    
    print(f'Profile created for user {user_id}')

channel.basic_consume(
    queue='orders.user_created',
    on_message_callback=on_user_created,
    auto_ack=True
)

Способ 6: ID передача через gRPC

Для синхронной communication между микросервисами:

# users_service.proto
syntax = "proto3";

package users;

service UserService {
    rpc GetUser(GetUserRequest) returns (User);
    rpc CreateUser(CreateUserRequest) returns (User);
}

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

message GetUserRequest {
    string user_id = 1;
}

message CreateUserRequest {
    string name = 1;
}

# users_service_pb2.py (сгенерировано)
# Service B (Orders) использует gRPC клиент

import grpc
from users import users_service_pb2, users_service_pb2_grpc

channel = grpc.aio.secure_channel(
    'users-service:50051',
    grpc.ssl_channel_credentials()
)
stub = users_service_pb2_grpc.UserServiceStub(channel)

async def create_order(user_id: str, amount: float):
    # Проверяем пользователя через gRPC
    request = users_service_pb2.GetUserRequest(user_id=user_id)
    user = await stub.GetUser(request)
    
    # Создаём заказ со ссылкой
    order = {
        'id': str(uuid4()),
        'user_id': user.id,  # user_id от Service A
        'amount': amount
    }
    
    return order

Способ 7: ID передача через заголовки (Correlation ID)

Для трассирования запроса через несколько сервисов:

# Service A
from uuid import uuid4
from fastapi import FastAPI, Header

app = FastAPI()

@app.get('/users/{user_id}')
def get_user(
    user_id: str,
    x_correlation_id: str = Header(default=str(uuid4()))
):
    # Логируем с correlation ID
    logger.info(f'Getting user {user_id}', extra={
        'correlation_id': x_correlation_id
    })
    
    # Service A делает запрос в Service B
    response = requests.get(
        f'https://orders-service/orders?user_id={user_id}',
        headers={
            'X-Correlation-ID': x_correlation_id,  # Пробрасываем дальше
            'X-User-ID': user_id
        }
    )
    
    return response.json()

Способ 8: Federated ID (для больших систем)

Пример: Google Workspace использует federated identity

# Каждая организация имеет свой ID namespace
class FederatedID:
    def __init__(self, organization_id: str, local_id: str):
        self.organization_id = organization_id
        self.local_id = local_id
    
    def __str__(self):
        return f"{self.organization_id}:{self.local_id}"
    
    @classmethod
    def from_string(cls, federated_id: str):
        org_id, local_id = federated_id.split(':')
        return cls(org_id, local_id)

# Использование
federated = FederatedID('org-123', 'user-456')
print(federated)  # org-123:user-456

# При передаче между сервисами
user_id = str(federated)  # "org-123:user-456"

Best Practices

# 1. Используй UUID для микросервисов
user_id = str(uuid4())

# 2. Передавай ID в JSON при REST API
response = {
    'id': user_id,
    'name': 'Alice'
}

# 3. Используй заголовки для metadata
headers = {
    'X-User-ID': user_id,
    'X-Correlation-ID': correlation_id
}

# 4. Валидируй ID на границе сервиса
from uuid import UUID

try:
    UUID(user_id)  # Проверяем формат UUID
except ValueError:
    raise HTTPException(400, 'Invalid user_id format')

# 5. Кэшируй информацию о пользователе, если часто запрашиваешь
from functools import lru_cache

@lru_cache(maxsize=1000)
def get_user_cached(user_id: str):
    return get_user(user_id)

Сравнение подходов

┌───────────────┬──────────┬─────────┬───────────┬──────────┐
│ Способ        │ Размер   │ Скорость│ Безопасн. │ Подходит │
├───────────────┼──────────┼─────────┼───────────┼──────────┤
│ UUID          │ 36 симв. │ Быстро  │ Хорошая   │ Все      │
│ Snowflake     │ 8 байт   │ Очень   │ Хорошая   │ BigData  │
│ Nanoid        │ 21 симв. │ Быстро  │ Хорошая   │ API      │
│ REST API      │ -        │ Среднее │ Depends   │ Синхр.   │
│ Message Queue │ -        │ Медленно│ Depends   │ Асинхр.  │
│ gRPC          │ -        │ Быстро  │ Depends   │ Internal │
└───────────────┴──────────┴─────────┴───────────┴──────────┘

Рекомендация: Используй UUID как основной формат ID, передавай через REST API/gRPC/Message Queue в зависимости от архитектуры.