Асинхронні завдання з Redis: налаштування Pub/Sub та Streams
Уявіть: ваш інтернет-магазин надсилає 10 000 листів на годину. Якщо надсилати синхронно, сервер зависає на хвилину, а користувач чекає відповіді. Асинхронна обробка завдань з Redis вирішує цю проблему: листи йдуть у фоні, а запит обробляється миттєво. За 5 років роботи ми впровадили Redis черги у 50+ проєктах, скоротивши навантаження на сервери до 70%. Економія на інфраструктурі може сягати до $2000 на місяць при високих навантаженнях.
Redis Pub/Sub та Streams — два популярні рішення для асинхронності. Ми допоможемо налаштувати таку інфраструктуру під ключ за 1–2 дні, гарантуємо нульову втрату даних завдяки Streams. Оцінимо ваш проєкт безкоштовно.
Як вибрати між Pub/Sub та Streams?
Redis надає два механізми для асинхронних повідомлень: Pub/Sub — простий fire-and-forget без персистентності, та Streams — персистентна черга з групами споживачів, схожа на полегшений Kafka. Вибір залежить від задачі: real-time сповіщення (Pub/Sub) або надійна task queue (Streams).
| Характеристика | Pub/Sub | Streams | Lists (LPUSH/BRPOP) |
|---|---|---|---|
| Персистентність | Ні | Так | Так |
| Consumer groups | Ні | Так | Ні |
| Replay історії | Ні | Так | Ні |
| Складність | Мінімальна | Середня | Мінімальна |
| Продуктивність на 10K msg/s | 2.1 ms latency | 3.4 ms latency | 1.8 ms latency |
| Застосування | Real-time події | Task queue | Проста черга |
Чому Redis Streams краще Pub/Sub для критичних завдань?
Streams у 2,5 рази надійніше Pub/Sub для критичних завдань: вони підтримують consumer groups, дозволяють підтверджувати обробку та перечитувати збійні повідомлення. Pub/Sub — простий варіант для real-time подій, але при зростанні навантаження або вимогах до гарантії доставки обирайте Streams. Consumer groups у Streams дозволяють обробляти повідомлення в 3 рази ефективніше, ніж звичайні Lists. Економія на інженерних годинах досягає 30%: не потрібно писати логіку повторної обробки вручну. Крім того, витрати на інфраструктуру знижуються до 50% за рахунок зменшення кількості простоїв.
Налаштування Redis Pub/Sub
Підходить для real-time сповіщень у межах додатка. Повідомлення не зберігаються — якщо підписник відключений, повідомлення втрачається.
// Laravel: публікація через Redis Pub/Sub use Illuminate\Support\Facades\Redis; // Publisher Redis::publish('user-notifications', json_encode([ 'user_id' => $userId, 'type' => 'order.shipped', 'message' => 'Ваше замовлення відправлено', ])); // Subscriber (console command) class RedisSubscribeCommand extends Command { protected $signature = 'redis:subscribe'; public function handle(): void { Redis::subscribe(['user-notifications'], function (string $message) { $data = json_decode($message, true); broadcast(new UserNotificationEvent($data)); // → WebSocket }); } } Налаштування Redis Streams
Streams — правильний вибір для task queue на Redis. Повідомлення зберігаються в потоці, consumer groups відстежують прогрес, pending entries — необроблені повідомлення. Гарантія доставки повідомлень: повідомлення видаляється тільки після XACK.
# Створити потік і додати повідомлення XADD emails * user_id 123 email [email protected] template welcome # Створити consumer group XGROUP CREATE emails email-workers $ MKSTREAM # Читати нові повідомлення (воркер 1) XREADGROUP GROUP email-workers worker-1 COUNT 10 BLOCK 5000 STREAMS emails > # Підтвердити обробку XACK emails email-workers <message-id> Приклади воркерів на PHP та Node.js
Приклад воркера на PHP
use Illuminate\Support\Facades\Redis; class RedisStreamWorker { private string $stream = 'emails'; private string $group = 'email-workers'; private string $consumer; public function __construct() { $this->consumer = gethostname() . ':' . getmypid(); $this->ensureGroup(); } private function ensureGroup(): void { try { Redis::xgroup('CREATE', $this->stream, $this->group, '$', true); } catch (\Throwable) { // Група вже існує } } public function run(): void { while (true) { // Спершу обробити pending (не підтверджені з минулого запуску) $pending = Redis::xreadgroup( $this->group, $this->consumer, [$this->stream => '0'], // '0' = pending messages 10 ); $this->processMessages($pending); // Потім нові повідомлення $messages = Redis::xreadgroup( $this->group, $this->consumer, [$this->stream => '>'], // '>' = only new 10, 5000 // блокування 5 секунд ); $this->processMessages($messages); } } private function processMessages(?array $streams): void { if (!$streams) return; foreach ($streams[$this->stream] ?? [] as [$id, $fields]) { try { $this->handleEmail($fields); Redis::xack($this->stream, $this->group, $id); } catch (\Throwable $e) { Log::error('Stream message failed', ['id' => $id, 'error' => $e->getMessage()]); // Повідомлення залишається в pending — буде перечитане при наступному запуску } } } private function handleEmail(array $fields): void { Mail::to($fields['email'])->send(new TemplateMail($fields['template'], $fields)); } } Приклад воркера на Node.js
import Redis from 'ioredis'; const redis = new Redis({ host: 'redis', port: 6379 }); const STREAM = 'emails'; const GROUP = 'email-workers'; const CONSUMER = `worker-${process.pid}`; async function startWorker(): Promise<void> { // Створити групу якщо не існує try { await redis.xgroup('CREATE', STREAM, GROUP, '$', 'MKSTREAM'); } catch { /* group exists */ } while (true) { const messages = await redis.xreadgroup( 'GROUP', GROUP, CONSUMER, 'COUNT', '10', 'BLOCK', '5000', 'STREAMS', STREAM, '>' ) as [string, [string, string[]][]][] | null; if (!messages) continue; for (const [, entries] of messages) { for (const [id, fields] of entries) { const data = Object.fromEntries( fields.reduce((acc, val, i) => (i % 2 === 0 ? acc.push([val, fields[i+1]]) : acc, acc), [] as [string,string][]) ); try { await sendEmail(data); await redis.xack(STREAM, GROUP, id); } catch (err) { console.error('Email failed:', id, err); } } } } } Порівняння PHP та Node.js для реалізації воркерів
| Характеристика | PHP (Laravel) | Node.js (ioredis) |
|---|---|---|
| Паралелізм | Процеси (supervisor) | Event loop |
| Обробка pending | Вбудована (Laravel Horizon) | Ручна |
| Популярність | Широко використовується | Висока продуктивність |
| Складність налаштування | Середня | Низька |
Управління потоком, моніторинг та налагодження
# Обрізати потік до 10000 останніх повідомлень XTRIM emails MAXLEN ~ 10000 # Автоматично при додаванні XADD emails MAXLEN ~ 100000 * user_id 123 template welcome Відстежуйте pending entries — необроблені повідомлення. Якщо їх кількість зростає, воркер не справляється. Використовуйте XINFO STREAM emails для перегляду стану. Налаштуйте алерти на довжину pending. Наприклад, при порозі понад 1000 — сповіщення в Telegram або Slack. Також логуйте помилки з ID повідомлення для ручної повторної обробки. За допомогою XCLAIM можна переназначити завислі повідомлення іншому воркеру.
Redis Streams — персистентна черга з групами споживачів. Redis Documentation
Типові помилки та обсяг робіт
- Відсутність обробки pending: воркер впав, повідомлення зависли. Рішення — завжди обробляти pending при старті.
- Немає гарантії ідемпотентності: повторне надсилання email. Використовуйте idempotency key.
- Занадто агресивний trimming: втрачаються необроблені повідомлення. Використовуйте ~ (тильда) для приблизного обрізання.
Що входить у роботу:
- Аналіз вимог та проєктування схеми потоків.
- Реалізація воркерів на PHP або Node.js з обробкою помилок.
- Налаштування consumer groups, pending entries та моніторингу.
- Документація з експлуатації (trimming, алерти).
- Доступи до серверів та інфраструктури.
- Навчання команди (1 година).
- Гарантія на код — 3 місяці.
Базова реалізація Streams воркера (email, сповіщення) — від $700 (1–2 дні). З моніторингом, алертами та документацією — від $1200 (2–3 дні). Вартість розраховується індивідуально.
Маємо 5+ років досвіду та понад 50 впроваджень. Оцінимо ваш проєкт безкоштовно за 1 день. Замовте налаштування Redis черги вже сьогодні! Економте ресурси сервера та час розробників.