Что такое Redis Pub/Sub?

Redis Pub/Sub

Redis Pub/Sub (Publish/Subscribe) — это система обмена сообщениями в реальном времени, реализованная в Redis. Она позволяет отправлять сообщения на каналы (channels), которые слушают подписчики (subscribers). Это асинхронный паттерн, полезный для построения реактивных систем и real-time приложений.

Основные концепции

Publisher (Издатель) — процесс или приложение, которое отправляет сообщение на канал. Издатель не нуждается в информации о подписчиках:

import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Отправляем сообщение на канал
redis_client.publish('user:notifications', 'Welcome, John!')

Subscriber (Подписчик) — процесс, который слушает сообщения на одном или нескольких каналах:

import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
pubsub = redis_client.pubsub()

# Подписываемся на канал
pubsub.subscribe('user:notifications')

# Слушаем сообщения
for message in pubsub.listen():
    if message['type'] == 'message':
        print(f"Received: {message['data']}")

Channel (Канал) — очередь сообщений, через которую издатель отправляет сообщения подписчикам. Каналы не хранятся в памяти Redis постоянно — это просто названия для маршрутизации:

# Отправляем на разные каналы
redis_client.publish('notifications:emails', 'Email sent')
redis_client.publish('notifications:sms', 'SMS sent')
redis_client.publish('notifications:push', 'Push sent')

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

import redis
import json
from typing import Callable

class NotificationPublisher:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
    
    def notify_user(self, user_id: int, message: str, notification_type: str = 'info') -> None:
        # Формируем сообщение
        notification = {
            'user_id': user_id,
            'message': message,
            'type': notification_type
        }
        # Публикуем на канал
        channel = f'user:{user_id}:notifications'
        self.redis.publish(channel, json.dumps(notification))

class NotificationSubscriber:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
        self.pubsub = self.redis.pubsub()
    
    def listen_for_user(self, user_id: int, callback: Callable) -> None:
        # Подписываемся на канал пользователя
        channel = f'user:{user_id}:notifications'
        self.pubsub.subscribe(channel)
        
        print(f"Listening for notifications on {channel}...")
        for message in self.pubsub.listen():
            if message['type'] == 'message':
                notification = json.loads(message['data'])
                callback(notification)

# Использование
publisher = NotificationPublisher()
subscriber = NotificationSubscriber()

def handle_notification(notif):
    print(f"User {notif['user_id']} received: {notif['message']}")

# В отдельном потоке слушаем
import threading
thread = threading.Thread(
    target=subscriber.listen_for_user, 
    args=(123, handle_notification)
)
thread.start()

# В основном потоке отправляем
publisher.notify_user(123, 'New comment on your post')

Pattern Subscribe (подписка на шаблоны)

Mожно подписаться не на конкретный канал, а на шаблон каналов:

import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
pubsub = redis_client.pubsub()

# Подписываемся на все каналы, начинающиеся с "user:"
pubsub.psubscribe('user:*')

for message in pubsub.listen():
    if message['type'] == 'pmessage':
        print(f"Channel: {message['channel']}, Data: {message['data']}")

Преимущества Redis Pub/Sub

1. Real-time доставка — сообщения доставляются мгновенно
2. Простота — минимум boilerplate кода
3. Масштабируемость — поддерживает большое количество подписчиков
4. Decoupling — издатель и подписчик не знают друг о друге
5. Гибкость — поддержка шаблонов и динамических каналов

Недостатки Redis Pub/Sub

1. Нет persistence — если подписчик отключен, он теряет сообщения
2. Нет гарантии доставки — сообщения не сохраняются в Redis
3. Нет истории — нельзя получить старые сообщения
4. Нет подтверждения доставки — издатель не знает, получено ли сообщение

Альтернативы: Redis Streams

Для случаев, когда нужна гарантия доставки, используются Redis Streams:

# Отправляем в stream
redis_client.xadd('notifications', {'user_id': '123', 'message': 'New message'})

# Читаем из stream
for message_id, data in redis_client.xread({'notifications': '0'}):
    print(f"ID: {message_id}, Data: {data}")

Практическое применение

1. WebSocket уведомления — real-time обновления для браузеров
2. Chat приложения — доставка сообщений между пользователями
3. Event-driven системы — распространение событий по микросервисам
4. Live нотификации — активность друзей, обновления статусов
5. Спортивные результаты — трансляция live событий

Redis Pub/Sub — это лёгкое и мощное решение для real-time коммуникации в распределённых системах.