В чём разница между request-reply и pub-sub моделями?

Разница между Request-Reply и Pub-Sub моделями: архитектурный выбор

Request-Reply и Pub-Sub — две фундаментальные паттерна асинхронной коммуникации в распределённых системах. Они решают разные проблемы и требуют разных инструментов, транспорта и мышления.

1. Основные характеристики

Request-Reply (Синхронная коммуникация)

Клиент отправляет запрос и ждёт ответ. Синхронно. Точка-к-точке.

Характеристики:

• Синхронность: Отправитель ждёт ответ
• Связность: Tight coupling
• Гарантии: Обычно есть таймаут
• Примеры: REST API, gRPC, Socket

Pub-Sub (Асинхронная коммуникация)

Продюсер публикует событие, не ждёт ответа. Консьюмеры подписаны на топики.

Характеристики:

• Асинхронность: Продюсер не ждёт
• Связность: Loose coupling
• Гарантии: Доставка, порядок
• Примеры: RabbitMQ, Kafka, Redis

2. Таблица сравнения

3. Request-Reply: Пример

import requests

def call_service():
    # Синхронно ждёт ответ
    response = requests.post(
        "http://service:8000/calculate",
        json={"a": 5, "b": 3},
        timeout=5
    )
    result = response.json()["result"]
    return result  # Получил результат

Проблемы:

• Клиент блокируется на таймаут
• Если сервис медленный - клиент ждёт
• Масштабируемость: нужно множество соединений

4. Pub-Sub: Пример с RabbitMQ

import pika
import json

# Продюсер
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.exchange_declare(exchange="events", exchange_type="topic")

# Публикует событие и СРАЗУ возвращает управление
channel.basic_publish(
    exchange="events",
    routing_key="order.created",
    body=json.dumps({"order_id": 123})
)
print("Event published!")  # Выполнится сразу

# Консьюмер
def email_consumer():
    def callback(ch, method, properties, body):
        order = json.loads(body)
        print(f"Sending email for order {order['order_id']}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    
    channel.basic_consume(
        queue="email_queue",
        on_message_callback=callback
    )
    channel.start_consuming()  # Слушает асинхронно

5. Практический пример: Обработка заказов

Неправильно: Request-Reply

from fastapi import FastAPI

app = FastAPI()

@app.post("/orders")
async def create_order(order: Order):
    # Клиент ждёт пока выполнится ВСЁ
    order_id = db.save_order(order)  # 100мс
    email_service.send(order.email)  # 500мс - ЗАВИСАНИЕ
    inventory_service.update(order.items)  # 200мс
    shipping_service.create(order_id)  # 300мс
    # Итого: клиент ждёт 1.1 сек
    # Если один сервис падает - весь заказ падает
    return {"order_id": order_id}

Правильно: Pub-Sub

from fastapi import FastAPI

app = FastAPI()

@app.post("/orders")
async def create_order(order: Order):
    # 1. Сохранить заказ
    order_id = db.save_order(order)  # 100мс
    
    # 2. Публиковать событие
    await message_broker.publish(
        "order.created",
        {"order_id": order_id, **order.dict()}
    )  # 1мс
    
    # Клиент получает ответ сразу! 101мс вместо 1.1 сек
    return {"order_id": order_id}

# Консьюмеры обрабатывают асинхронно

async def email_consumer():
    async for order in broker.subscribe("order.created"):
        await email_service.send(order.email)

async def inventory_consumer():
    async for order in broker.subscribe("order.created"):
        await inventory_service.update(order.items)

async def shipping_consumer():
    async for order in broker.subscribe("order.created"):
        await shipping_service.create(order.order_id)

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

• Клиент получает ответ быстро
• Все операции выполняются асинхронно
• Если email падает - заказ уже создан
• Легко добавить нового консьюмера
• Масштабируется горизонтально

6. Пример с Kafka

from kafka import KafkaProducer, KafkaConsumer
import json

# Продюсер
producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

# Публикует и НЕ ждёт
producer.send("orders", {"order_id": 123})
print("Published!")  # Выполнится сразу

# Консьюмер
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers="localhost:9092",
    group_id="order_processors",
    value_deserializer=lambda m: json.loads(m.decode("utf-8"))
)

for message in consumer:
    order = message.value
    print(f"Processing order {order['order_id']}")
    # Обрабатывает асинхронно
    # Брокер гарантирует доставку

7. Когда использовать что?

Request-Reply подходит для:

• Синхронные операции где нужен результат (GET user)
• Транзакционные операции (платёж)
• Быстрые операции (weather API)
• Когда нужна немедленная обратная связь

Pub-Sub подходит для:

• События интересующие много систем
• Асинхронная обработка
• Высокая нагрузка (1000+ событий/сек)
• Если консьюмер падает, сообщения сохраняются
• Loose coupling между системами

8. Гибридный подход

@app.post("/orders")
async def create_order(order: Order):
    # Validation (синхронно) - Request-Reply
    if not is_valid_order(order):
        return {"error": "Invalid"}  # Ошибка нужна сразу
    
    # Сохранение (синхронно) - Request-Reply
    order_id = db.save_order(order)  # Нужен ID
    
    # Обработка (асинхронно) - Pub-Sub
    await broker.publish("order.created", {"order_id": order_id})
    # Не ждём, результат не важен
    
    return {"order_id": order_id, "status": "processing"}

Резюме

Request-Reply:

• Клиент ждёт ответ
• Tight coupling
• Для быстрых, синхронных операций
• Сложнее масштабировать

Pub-Sub:

• Продюсер не ждёт
• Loose coupling
• Для событий и асинхронной обработки
• Масштабируется горизонтально
• Отказоустойчив

Выбор:

• Нужен результат срочно? → Request-Reply
• События интересуют много систем? → Pub-Sub
• Высокая нагрузка? → Pub-Sub
• Нужна отказоустойчивость? → Pub-Sub