Основы работы с Redis Pub/Sub

Лёгкий messaging: каналы, publish/subscribe, паттерны. Когда Redis подходит для очередей, а когда нет.

#Redis как брокер сообщений

Redis — это не только key-value хранилище, но и простой брокер сообщений с механизмом Pub/Sub.

Особенности Redis Pub/Sub:

Без персистентности: сообщения не хранятся, доставляются только активным подписчикам
Минимальная задержка: <1ms, быстрее RabbitMQ и Kafka
Простая модель: каналы без сложной маршрутизации
Fire-and-forget: нет acknowledgment, retry, DLQ

Когда использовать:

✅ Real-time уведомления (чаты, онлайн-статусы, лайвы)
✅ Кэш + pub/sub в одном (Redis как хранилище + messaging)
✅ Простые сценарии без требований к надёжности
❌ Критичные данные (платежи, заказы) — сообщения теряются
❌ Очереди задач с гарантией доставки
❌ История событий (event sourcing)

#Архитектура Redis Pub/Sub

┌─────────────┐
│  Publisher  │
└──────┬──────┘
       │ PUBLISH notifications "Hello!"
       ▼
┌─────────────────┐
│     Redis       │
│  (Message Hub)  │
└──────┬──────────┘
       │
       ├──────────────┐
       │              │
       ▼              ▼
┌─────────────┐ ┌─────────────┐
│ Subscriber  │ │ Subscriber  │
│   (Email)   │ │   (Push)    │
└─────────────┘ └─────────────┘

Модель:

Publisher публикует сообщение в канал
Redis доставляет сообщение всем активным подписчикам канала
Подписчики получают сообщение немедленно
Если подписчик недоступен — сообщение теряется

#Работа с Redis Pub/Sub в FastStream

#Конфигурация

from faststream import FastStream
from faststream.redis import RedisBroker

broker = RedisBroker("redis://localhost:6379")

app = FastStream(broker)

#Publisher

@broker.publisher("notifications")
async def send_notification(user_id: int, message: str):
    await broker.publish(
        {"user_id": user_id, "message": message},
        "notifications"
    )

#Subscriber

from pydantic import BaseModel

class Notification(BaseModel):
    user_id: int
    message: str

@broker.subscriber("notifications")
async def handle_notification(notif: Notification):
    print(f"User {notif.user_id}: {notif.message}")

#Паттерны Redis Pub/Sub

#1. Простой Pub/Sub

Базовая публикация-подписка:

# Publisher
await broker.publish({"event": "user.login"}, "events")

# Subscriber
@broker.subscriber("events")
async def handle_event(event: dict):
    ...

#2. Pattern Sub (подписка по паттерну)

Подписка на несколько каналов по паттерну:

# Публикация в разные каналы
await broker.publish({"msg": "error"}, "logs.error")
await broker.publish({"msg": "warning"}, "logs.warning")
await broker.publish({"msg": "info"}, "logs.info")

# Подписка на все логи
@broker.subscriber("logs.*")  # Паттерн
async def handle_all_logs(msg: dict):
    ...

# Подписка только на ошибки
@broker.subscriber("logs.error")
async def handle_errors(msg: dict):
    ...

Паттерны:

logs.* — logs.error, logs.warning (один уровень)
logs.** — logs.error, logs.error.db, logs.error.db.sql (любая вложенность)
user.*.events — user.123.events, user.456.events

#3. Request-Reply (RPC)

Синхронный вызов через Redis:

# Server
@broker.subscriber("calculate", reply_to="results")
async def handle_calc(data: dict):
    result = data["a"] + data["b"]
    return {"result": result}

# Client
response = await broker.request(
    {"a": 2, "b": 3},
    "calculate",
    timeout=5.0
)
print(response)  # {"result": 5}

#4. Stream (персистентность)

Redis Streams — персистентная альтернатива Pub/Sub:

# Публикация в stream
await broker.publish(
    {"order_id": 123},
    "orders",
    stream=True  # Использовать Redis Streams
)

# Потребление из stream
@broker.subscriber(
    "orders",
    stream=True,
    group="order-processors",  # Consumer group
    consumer="worker-1"
)
async def process_order(order: dict):
    ...

Отличия Streams от Pub/Sub:

Сообщения хранятся в логе (как Kafka)
Поддерживаются consumer groups
Есть acknowledgment (XACK)
Можно читать старые сообщения (XREAD)

#Продвинутые возможности

#Batch publishing

Публикация нескольких сообщений в одном запросе:

from faststream.redis import RedisBatch

async with RedisBatch(broker) as batch:
    batch.publish({"id": 1}, "orders")
    batch.publish({"id": 2}, "orders")
    batch.publish({"id": 3}, "orders")
    # Один RTT вместо трёх

#List-based queues

Очереди на основе списков Redis (альтернатива Pub/Sub):

# Producer: LPUSH
await broker.connection.lpush("tasks", json.dumps(task))

# Consumer: BRPOP (blocking)
@broker.subscriber("tasks", list=True)
async def process_task(task: dict):
    ...

Преимущества перед Pub/Sub:

Сообщения хранятся в списке
Потребитель получает сообщение, даже если опубликовано до его старта
Поддерживается blocking pop (BRPOP)

#Pub/Sub с метаданными

Передача метаданных через заголовки:

await broker.publish(
    {"data": "..."},
    "channel",
    headers={"priority": "high", "trace_id": "abc123"}
)

@broker.subscriber("channel")
async def handle(msg: dict, message: RedisMessage):
    print(f"Priority: {message.headers.get('priority')}")

#Пример: система уведомлений

from faststream import FastStream
from faststream.redis import RedisBroker
from pydantic import BaseModel
from enum import Enum

class NotificationType(str, Enum):
    EMAIL = "email"
    PUSH = "push"
    SMS = "sms"

class Notification(BaseModel):
    user_id: int
    type: NotificationType
    message: str

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)

# Publisher
@broker.publisher("notifications")
async def notify_user(notif: Notification):
    await broker.publish(notif.model_dump(), "notifications")

# Email subscriber
@broker.subscriber("notifications")
async def send_email(notif: Notification):
    if notif.type == NotificationType.EMAIL:
        print(f"Email to {notif.user_id}: {notif.message}")

# Push subscriber
@broker.subscriber("notifications")
async def send_push(notif: Notification):
    if notif.type == NotificationType.PUSH:
        print(f"Push to {notif.user_id}: {notif.message}")

# Логирование всех уведомлений (pattern sub)
@broker.subscriber("notifications.*")
async def log_notification(notif: Notification):
    print(f"Logged: {notif}")

#Мониторинг Redis Pub/Sub

#Redis CLI

# Просмотр активных подписчиков
redis-cli PUBSUB NUMSUB notifications
# notifications: 5 (5 подписчиков)

# Просмотр подписок по паттерну
redis-cli PUBSUB NUMPAT logs.*

# Мониторинг в реальном времени
redis-cli MONITOR

#Метрики

Redis экспортирует метрики через INFO:

pubsub_channels — активные каналы
pubsub_patterns — активные паттерны
connected_clients — подключённые клиенты

#Сравнение: Pub/Sub vs Streams vs Lists

Характеристика	Pub/Sub	Streams	Lists
Персистентность	Нет	Да	Да
Доставка	Всем подписчикам	Consumer groups	Один потребитель
Ack	Нет	Да (XACK)	Нет (но можно эмулировать)
Replay	Нет	Да (XREAD)	Да (LRANGE)
Задержка	<1ms	~1ms	~1ms
Сценарий	Real-time уведомления	Event sourcing, лог	Очереди задач

┌─────────────┐ │ Publisher │ └──────┬──────┘ │ PUBLISH notifications "Hello!" ▼ ┌─────────────────┐ │ Redis │ │ (Message Hub) │ └──────┬──────────┘ │ ├──────────────┐ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Subscriber │ │ Subscriber │ │ (Email) │ │ (Push) │ └─────────────┘ └─────────────┘

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

Pub/Sub

Streams

Lists

Персистентность

Нет

Да

Доставка

Всем подписчикам

Consumer groups

Один потребитель

Ack

Нет

Да (XACK)

Нет (но можно эмулировать)

Replay

Нет

Да (XREAD)

Да (LRANGE)

Задержка

<1ms

~1ms

Сценарий

Real-time уведомления

Event sourcing, лог

Очереди задач