Настройка Redis для Pub/Sub уведомлений

Настройка Redis для Pub/Sub уведомлений

Redis Pub/Sub — механизм доставки сообщений в режиме «fire and forget». Брокер не хранит историю: если подписчик не слушает в момент публикации, сообщение теряется. Это отличает его от Redis Streams и делает подходящим для коротких уведомлений в реальном времени — оповещений о событиях, инвалидации кешей, синхронизации состояния между несколькими экземплярами сервиса.

Когда Pub/Sub, а когда очередь

Pub/Sub решает задачу broadcast: один издатель, несколько подписчиков, все получают одно и то же сообщение одновременно. Если нужна гарантированная доставка, история сообщений или группы потребителей — Redis Streams или RabbitMQ.

Типичный кейс для сайта: пользователь выполнил действие → бэкенд публикует событие в канал → WebSocket-сервер получает событие и рассылает push всем подключённым клиентам соответствующей «комнаты». Без Pub/Sub горизонтальное масштабирование WebSocket-серверов невозможно: каждый экземпляр знает только о своих соединениях.

Базовая настройка

Redis из коробки поддерживает Pub/Sub без конфигурации. Но стоит выставить несколько параметров в redis.conf:

# Лимит памяти — критично, если Redis используется и как кеш
maxmemory 512mb
maxmemory-policy allkeys-lru

# Количество баз — для prod достаточно одной логической БД
databases 16

# Отключить persistence для чистого pub/sub-брокера
save ""
appendonly no

Для production Redis должен быть в режиме Sentinel или Cluster. Pub/Sub в Cluster работает с ограничением: сообщения распространяются только внутри шарда, если не использовать SPUBLISH/SSUBSCRIBE (Sharded Pub/Sub, Redis 7+).

Реализация на Node.js с ioredis

import Redis from 'ioredis';

const publisher = new Redis({ host: 'redis', port: 6379 });
const subscriber = new Redis({ host: 'redis', port: 6379 });

// Подписчик — отдельное соединение, заблокированное на listen
subscriber.subscribe('notifications:user:*', (err, count) => {
  if (err) throw err;
  console.log(`Subscribed to ${count} channels`);
});

subscriber.on('pmessage', (pattern, channel, message) => {
  // channel = "notifications:user:42"
  const userId = channel.split(':')[2];
  const payload = JSON.parse(message);
  broadcastToUser(userId, payload);
});

// Публикация из любого другого места
async function notifyUser(userId: string, event: object) {
  const channel = `notifications:user:${userId}`;
  const count = await publisher.publish(channel, JSON.stringify(event));
  // count — число подписчиков, получивших сообщение
  return count;
}

Важно: publisher и subscriber — разные соединения. После вызова subscribe/psubscribe соединение переходит в специальный режим и принимает только команды SUBSCRIBE, UNSUBSCRIBE, PING, RESET, QUIT.

Pattern-matching подписки

PSUBSCRIBE поддерживает глобы: * (любые символы), ? (один символ), [chars] (набор символов). Полезно для подписки на целый класс каналов:

// Все события конкретного тенанта
subscriber.psubscribe('tenant:acme:*');

// Конкретный тип события для всех пользователей
subscriber.psubscribe('*:message:new');

Интеграция с WebSocket (Socket.io)

Классическая схема масштабирования Socket.io через Redis Adapter:

import { createServer } from 'http';
import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

const httpServer = createServer();
const io = new Server(httpServer);

const pubClient = createClient({ url: 'redis://redis:6379' });
const subClient = pubClient.duplicate();

await Promise.all([pubClient.connect(), subClient.connect()]);
io.adapter(createAdapter(pubClient, subClient));

// Теперь io.to('room').emit() работает на всех экземплярах сервера
httpServer.listen(3000);

Redis Adapter использует Pub/Sub внутри: при вызове io.to('room').emit() на одном экземпляре, команда публикуется в Redis-канал, и все остальные экземпляры получают её и рассылают своим подключённым клиентам из этой комнаты.

Интеграция с Laravel

Laravel Echo Server или Soketi — стандартный путь. Но можно напрямую через predis или phpredis:

// Публикация события из Laravel
use Illuminate\Support\Facades\Redis;

Redis::publish('notifications:user:' . $userId, json_encode([
    'type' => 'order.status_changed',
    'orderId' => $order->id,
    'status' => $order->status,
    'timestamp' => now()->toISOString(),
]));

Для Laravel Broadcasting с Redis:

// config/broadcasting.php
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => '{default}',
        'retry_after' => 90,
    ],
],

# Запуск Laravel Echo Server
npx laravel-echo-server start
# или Soketi (более современная альтернатива)
soketi start --config=soketi.json

Мониторинг

# Просмотр активных Pub/Sub каналов
redis-cli PUBSUB CHANNELS "*"

# Число подписчиков на канале
redis-cli PUBSUB NUMSUB notifications:user:42

# Паттерн-подписки
redis-cli PUBSUB NUMPAT

# Мониторинг всех команд в реальном времени (осторожно на prod)
redis-cli MONITOR

В Redis Exporter для Prometheus метрики Pub/Sub доступны через redis_connected_slaves и custom-скрипты. Важная метрика — instantaneous_ops_per_sec: если она резко растёт при Pub/Sub нагрузке, проверьте размер сообщений и частоту публикации.

Ограничения и подводные камни

Потеря сообщений при реконнекте. Если подписчик отключился и переподключился, пропущенные сообщения не восстанавливаются. Для критичных уведомлений — Redis Streams с consumer groups или хранение последних событий в отдельном ключе.

Нет подтверждения доставки. PUBLISH возвращает число получателей, но не гарантирует, что они обработали сообщение. Для at-least-once доставки нужна очередь.

Нагрузка на CPU при большом числе паттернов. PSUBSCRIBE сопоставляет каждое опубликованное сообщение со всеми зарегистрированными паттернами. При 10 000+ паттернов это становится заметным.

Сроки

Базовая интеграция Redis Pub/Sub с Socket.io или Soketi на существующем проекте — 1–2 дня. Полноценная система уведомлений с персонализацией каналов, обработкой переподключений и мониторингом — 4–6 дней.