Redis Streams: event bus, rate limiting, audit log без Kafka

Redis — это не только кэш

Большинство разработчиков используют Redis для двух вещей: кэширование и очереди (через Sidekiq/BullMQ). Но Redis давно перерос роль «быстрого key-value хранилища». Pub/Sub для real-time уведомлений, Sorted Sets для лидербордов, HyperLogLog для уникальных счётчиков, Streams для event-driven архитектуры — всё это встроено и работает со скоростью in-memory базы.

Redis Streams — самая недооценённая структура данных. Появилась в Redis 5.0, в 7.x стала по-настоящему зрелой. По сути это append-only log (как Kafka), но с API Redis и без необходимости разворачивать кластер из 6 нод с ZooKeeper.

В моих проектах Streams заменил три вещи: кастомный Pub/Sub для real-time уведомлений, polling-based синхронизацию между сервисами и отдельную очередь для event sourcing. Один Redis, одна структура данных — три задачи.

Что такое Redis Streams

Концепция

Stream — это append-only лог с ID на каждую запись. Записи состоят из пар ключ-значение. Записи упорядочены по времени и никогда не изменяются после добавления.

Stream: orders
┌──────────────────┬────────────────────────────────────┐
│ ID               │ Fields                             │
├──────────────────┼────────────────────────────────────┤
│ 1716000000000-0  │ action=created order_id=1001       │
│ 1716000000001-0  │ action=paid order_id=1001          │
│ 1716000000002-0  │ action=created order_id=1002       │
│ 1716000000003-0  │ action=shipped order_id=1001       │
│ 1716000000004-0  │ action=cancelled order_id=1002     │
└──────────────────┘────────────────────────────────────┘

ID формат: <timestamp_ms>-<sequence>. Генерируется автоматически (*) или задаётся вручную. Timestamp гарантирует хронологический порядок, sequence — уникальность внутри миллисекунды.

Базовые операции

# Добавить запись в stream
XADD orders * action created order_id 1001 amount 99.99
# "1716000000000-0"

# Добавить с автоматическим ID
XADD orders * action paid order_id 1001 payment_method card

# Прочитать все записи
XRANGE orders - +
# Прочитать последние 10
XRANGE orders - + COUNT 10

# Прочитать с определённого ID
XRANGE orders 1716000000001-0 +

# Прочитать в обратном порядке (последние N)
XREVRANGE orders + - COUNT 5

# Длина stream
XLEN orders

# Информация о stream
XINFO STREAM orders

Stream vs Pub/Sub vs List

Stream — единственная структура, которая сочетает персистентность, повторное чтение, consumer groups и acknowledgment.

Consumer Groups — параллельная обработка

Концепция

Consumer Group — это механизм распределения записей между несколькими потребителями. Каждая запись доставляется ровно одному consumer в группе (в отличие от Pub/Sub, где все получают всё).

Stream: orders
    │
    ├── Consumer Group: "payment-service"
    │       ├── Consumer: payment-1  ← получает запись A
    │       └── Consumer: payment-2  ← получает запись B
    │
    └── Consumer Group: "notification-service"
            ├── Consumer: notif-1    ← получает запись A
            └── Consumer: notif-2    ← получает запись B

Две группы читают один stream независимо. Внутри группы записи распределяются между consumer-ами.

Создание и чтение

# Создать consumer group
# $  — начать с новых записей
# 0  — начать с самого начала
XGROUP CREATE orders payment-service $ MKSTREAM
XGROUP CREATE orders notification-service 0 MKSTREAM

# Читать из группы (блокирующе)
# >  — только новые, ещё не доставленные записи
XREADGROUP GROUP payment-service payment-worker-1 COUNT 10 BLOCK 5000 STREAMS orders >

# Подтвердить обработку
XACK orders payment-service 1716000000000-0

# Посмотреть pending (необработанные) записи
XPENDING orders payment-service - + 10

# Claim — забрать «застрявшие» записи у упавшего consumer-а
# (записи, не подтверждённые более 60 секунд)
XAUTOCLAIM orders payment-service new-worker-1 60000 0-0 COUNT 10

Жизненный цикл записи

1. Producer: XADD orders * action created ...
   └── Запись добавлена в stream

2. Consumer: XREADGROUP GROUP payment-service worker-1 ... STREAMS orders >
   └── Запись доставлена worker-1, попала в pending list

3. Worker обрабатывает запись
   └── Бизнес-логика: создать платёж, отправить webhook

4. Worker: XACK orders payment-service <id>
   └── Запись подтверждена, удалена из pending list

5. (Если worker упал) Другой worker: XAUTOCLAIM ...
   └── Забирает «застрявшие» записи и обрабатывает

Практические паттерны

Event Bus между микросервисами

┌─────────────┐     ┌───────────────────┐     ┌──────────────────┐
│  Order API   │────▶│  Stream: events   │────▶│ Payment Service  │
│  (producer)  │     │                   │     │ (consumer group) │
└─────────────┘     │  order.created     │     └──────────────────┘
                    │  order.paid        │
┌─────────────┐     │  order.shipped     │     ┌──────────────────┐
│  User API    │────▶│  user.registered  │────▶│ Email Service    │
│  (producer)  │     │  user.updated     │     │ (consumer group) │
└─────────────┘     └───────────────────┘     └──────────────────┘

Один stream events — все сервисы пишут события. Каждый сервис-consumer создаёт свою group и читает только нужные события (фильтрация на стороне consumer).

Rate Limiting через Streams

import redis
import time

r = redis.Redis()

def is_rate_limited(user_id: str, limit: int = 100, window_seconds: int = 60) -> bool:
    stream_key = f"ratelimit:{user_id}"
    now = int(time.time() * 1000)
    window_start = now - (window_seconds * 1000)

    # Удалить записи старше окна
    r.xtrim(stream_key, minid=window_start)

    # Посчитать записи в окне
    current_count = r.xlen(stream_key)

    if current_count >= limit:
        return True

    # Добавить текущий запрос
    r.xadd(stream_key, {"ts": now}, maxlen=limit * 2)
    r.expire(stream_key, window_seconds * 2)

    return False

Преимущество перед обычным counter: видна история запросов, можно анализировать паттерны, sliding window без гонок.

Activity Feed / Timeline

# Добавить активность
XADD user:123:feed * type comment post_id 456 text "Great article!"
XADD user:123:feed * type like post_id 789
XADD user:123:feed * type follow target_user_id 321

# Последние 20 элементов фида
XREVRANGE user:123:feed + - COUNT 20

# Фид после определённой позиции (пагинация)
XREVRANGE user:123:feed (1716000000000-0 - COUNT 20

# Ограничить размер фида
XTRIM user:123:feed MAXLEN ~ 1000

MAXLEN ~ 1000 — приблизительная обрезка. ~ означает, что Redis может оставить чуть больше 1000 записей (для эффективности — не нужно обрезать каждый блок). Точная обрезка — MAXLEN 1000, но она медленнее.

Real-time аналитика

# Каждый pageview — запись в stream
XADD pageviews * url /products/123 user_id 456 referrer google device mobile

# Consumer агрегирует в реальном времени
# (каждые N записей или каждые N секунд)

# app/services/pageview_aggregator.rb
class PageviewAggregator
  STREAM = "pageviews"
  GROUP = "aggregator"
  CONSUMER = "agg-#{Process.pid}"
  BATCH_SIZE = 100

  def run
    ensure_group_exists

    loop do
      entries = redis.xreadgroup(GROUP, CONSUMER, STREAM, ">",
                                 count: BATCH_SIZE, block: 5000)
      next if entries.empty?

      entries.each do |_stream, messages|
        messages.each do |id, fields|
          process_pageview(fields)
          redis.xack(STREAM, GROUP, id)
        end
      end

      flush_aggregates
    end
  end

  private

  def process_pageview(fields)
    url = fields["url"]
    hour = Time.current.beginning_of_hour.iso8601

    @counts ||= Hash.new(0)
    @counts["#{hour}:#{url}"] += 1
  end

  def flush_aggregates
    return if @counts.nil? || @counts.empty?

    @counts.each do |key, count|
      hour, url = key.split(":", 2)
      redis.hincrby("pageview_stats:#{hour}", url, count)
    end

    @counts.clear
  end
end

Audit Log

# Каждое изменение — запись в stream
XADD audit:company:42 * \
  user_id 5 \
  action update \
  entity Employee \
  entity_id 123 \
  changes '{"salary": [50000, 55000], "position": ["Junior", "Middle"]}' \
  ip 192.168.1.100

# История изменений за последний час
XRANGE audit:company:42 1716000000000 +

# Все изменения конкретной сущности
# (фильтрация на стороне клиента — stream не поддерживает WHERE)

Stream как audit log — дешевле чем таблица в PostgreSQL для write-heavy сценариев. Можно асинхронно переносить в PostgreSQL для долгосрочного хранения.

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

Гем redis + producer

# config/initializers/redis.rb
REDIS = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))

# app/services/event_publisher.rb
class EventPublisher
  def self.publish(stream, event_type, data = {})
    REDIS.xadd(
      stream,
      { event_type: event_type, **data, published_at: Time.current.iso8601 },
      maxlen: ["~", 100_000]
    )
  end
end

# Использование в контроллере/сервисе
class OrderService
  def create(params)
    order = Order.create!(params)

    EventPublisher.publish("events:orders", "order.created",
      order_id: order.id,
      user_id: order.user_id,
      total: order.total.to_s
    )

    order
  end

  def pay(order, payment_method:)
    order.update!(status: :paid, paid_at: Time.current)

    EventPublisher.publish("events:orders", "order.paid",
      order_id: order.id,
      payment_method: payment_method
    )
  end
end

Consumer как фоновая задача

# app/jobs/stream_consumer_job.rb
class StreamConsumerJob < ApplicationJob
  STREAM = "events:orders"
  GROUP = "email-notifications"
  CONSUMER = "worker-#{Process.pid}"

  def perform
    ensure_group

    loop do
      results = REDIS.xreadgroup(GROUP, CONSUMER, STREAM, ">",
                                  count: 50, block: 5000)

      break if results.nil?

      results.each do |_stream, messages|
        messages.each do |id, fields|
          process_message(fields)
          REDIS.xack(STREAM, GROUP, id)
        rescue => e
          Rails.logger.error("Stream consumer error: #{e.message}")
        end
      end
    end
  end

  private

  def ensure_group
    REDIS.xgroup(:create, STREAM, GROUP, "$", mkstream: true)
  rescue Redis::CommandError => e
    raise unless e.message.include?("BUSYGROUP")
  end

  def process_message(fields)
    case fields["event_type"]
    when "order.created"
      OrderMailer.confirmation(fields["order_id"]).deliver_later
    when "order.paid"
      OrderMailer.receipt(fields["order_id"]).deliver_later
    when "order.shipped"
      OrderMailer.shipping_notification(fields["order_id"]).deliver_later
    end
  end
end

Consumer как systemd-сервис

# lib/tasks/streams.rake
namespace :streams do
  desc "Run stream consumer for email notifications"
  task consume_orders: :environment do
    consumer = StreamConsumerJob.new
    consumer.perform
  end
end

# /etc/systemd/system/stream-consumer.service
[Unit]
Description=Redis Stream Consumer - Email Notifications
After=network.target redis.service

[Service]
Type=simple
User=deploy
WorkingDirectory=/var/www/app
ExecStart=/usr/bin/bundle exec rails streams:consume_orders
Restart=always
RestartSec=5
Environment=RAILS_ENV=production

[Install]
WantedBy=multi-user.target

Интеграция с Node.js

import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);

// Producer
async function publishEvent(stream: string, data: Record<string, string>) {
  await redis.xadd(stream, '*', ...Object.entries(data).flat());
}

// Consumer с Consumer Group
async function consumeStream(
  stream: string,
  group: string,
  consumer: string,
  handler: (fields: Record<string, string>) => Promise<void>
) {
  // Создать группу если не существует
  try {
    await redis.xgroup('CREATE', stream, group, '$', 'MKSTREAM');
  } catch (e: any) {
    if (!e.message.includes('BUSYGROUP')) throw e;
  }

  while (true) {
    const results = await redis.xreadgroup(
      'GROUP', group, consumer,
      'COUNT', '50',
      'BLOCK', '5000',
      'STREAMS', stream, '>'
    );

    if (!results) continue;

    for (const [, messages] of results) {
      for (const [id, fields] of messages) {
        try {
          const data: Record<string, string> = {};
          for (let i = 0; i < fields.length; i += 2) {
            data[fields[i]] = fields[i + 1];
          }
          await handler(data);
          await redis.xack(stream, group, id);
        } catch (error) {
          console.error(`Failed to process ${id}:`, error);
        }
      }
    }
  }
}

// Использование
consumeStream('events:orders', 'analytics', 'analytics-1', async (fields) => {
  if (fields.event_type === 'order.created') {
    await updateDashboard(fields);
  }
});

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

import redis
import json
import time
import signal
import sys

r = redis.Redis.from_url("redis://localhost:6379/0")

class StreamConsumer:
    def __init__(self, stream: str, group: str, consumer: str):
        self.stream = stream
        self.group = group
        self.consumer = consumer
        self.running = True

        signal.signal(signal.SIGTERM, self._shutdown)
        signal.signal(signal.SIGINT, self._shutdown)

    def _shutdown(self, *args):
        self.running = False

    def ensure_group(self):
        try:
            r.xgroup_create(self.stream, self.group, id="$", mkstream=True)
        except redis.ResponseError as e:
            if "BUSYGROUP" not in str(e):
                raise

    def run(self, handler):
        self.ensure_group()

        while self.running:
            results = r.xreadgroup(
                self.group, self.consumer,
                {self.stream: ">"},
                count=50, block=5000
            )

            if not results:
                continue

            for stream_name, messages in results:
                for msg_id, fields in messages:
                    try:
                        decoded = {k.decode(): v.decode() for k, v in fields.items()}
                        handler(decoded)
                        r.xack(self.stream, self.group, msg_id)
                    except Exception as e:
                        print(f"Error processing {msg_id}: {e}")

# Использование
def handle_order_event(fields):
    if fields["event_type"] == "order.created":
        send_confirmation_email(fields["order_id"])

consumer = StreamConsumer("events:orders", "email-service", "email-worker-1")
consumer.run(handle_order_event)

Управление памятью и retention

MAXLEN — ограничение по количеству

# Точное ограничение (каждый XADD проверяет и обрезает)
XADD mystream MAXLEN 10000 * key value

# Приблизительное ограничение (быстрее, обрезает блоками)
XADD mystream MAXLEN ~ 10000 * key value

MINID — ограничение по времени

# Удалить записи старше 24 часов
# (ID = timestamp в миллисекундах)
XADD mystream MINID ~ 1716000000000 * key value

# Или через XTRIM
XTRIM mystream MINID ~ $(date -d '24 hours ago' +%s)000

Автоматическая очистка через cron

# Каждый час: удалить записи старше 7 дней
0 * * * * redis-cli XTRIM events:orders MINID ~ $(date -d '7 days ago' +%s%3N)

# lib/tasks/redis_maintenance.rake
namespace :redis do
  desc "Trim old stream entries"
  task trim_streams: :environment do
    streams = %w[events:orders events:users events:payments]
    cutoff = 7.days.ago.to_i * 1000 # миллисекунды

    streams.each do |stream|
      before = REDIS.xlen(stream)
      REDIS.xtrim(stream, minid: ["~", cutoff.to_s])
      after = REDIS.xlen(stream)
      puts "#{stream}: #{before} → #{after} (trimmed #{before - after})"
    end
  end
end

Мониторинг потребления памяти

# Размер stream в записях
XLEN events:orders

# Детальная информация
XINFO STREAM events:orders

# Информация о consumer groups
XINFO GROUPS events:orders

# Информация о consumer-ах в группе
XINFO CONSUMERS events:orders payment-service

# Общее использование памяти Redis
INFO memory
# used_memory_human: 256.00M
# used_memory_dataset: 180.00M

# Память конкретного ключа
MEMORY USAGE events:orders
# (integer) 52428800  — ~50 MB

Redis Streams vs Kafka

Когда Redis Streams достаточно

• Менее 100K сообщений в секунду
• Допустима потеря нескольких сообщений при крэше (AOF everysec)
• Не нужна exactly-once семантика
• Уже есть Redis в инфраструктуре
• Команда без Kafka-экспертизы
• Простой event bus между 2-5 сервисами

Когда нужен Kafka

• Более 100K сообщений в секунду
• Гарантированная персистентность (финансовые транзакции)
• Exactly-once обработка
• Event sourcing с долгосрочным хранением
• 10+ consumer-ов на один топик
• Compliance-требования (audit trail)

Мониторинг Redis Streams

Prometheus + redis_exporter

# docker-compose.yml
services:
  redis-exporter:
    image: oliver006/redis_exporter:v1.60.0
    ports:
      - "9121:9121"
    environment:
      REDIS_ADDR: redis://redis:6379

Ключевые метрики:

# Длина stream (количество записей)
redis_stream_length{stream="events:orders"}

# Количество групп на stream
redis_stream_groups{stream="events:orders"}

# Pending записи (не подтверждённые)
redis_stream_group_pending{stream="events:orders",group="payment-service"}

# Consumer lag (разница между последней записью и позицией consumer-а)
redis_stream_group_last_delivered_id{stream="events:orders",group="payment-service"}

Алерты

# Prometheus alerting rules
groups:
  - name: redis-streams
    rules:
      - alert: StreamPendingHigh
        expr: redis_stream_group_pending > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High pending count in stream {{ $labels.stream }}"

      - alert: StreamConsumerLag
        expr: redis_stream_length - redis_stream_group_last_delivered_id > 10000
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Consumer lag > 10K in {{ $labels.stream }}/{{ $labels.group }}"

Итоговый чеклист

Перед использованием

• [ ] Определить паттерн: event bus, activity feed, rate limiting, audit log
• [ ] Оценить throughput: < 100K msg/sec → Redis Streams, > 100K → Kafka
• [ ] Оценить требования к durability: допустима ли потеря 1-2 секунд данных?
• [ ] Redis уже в инфраструктуре? Если нет — оценить стоит ли добавлять

Проектирование

• [ ] Именование streams: events:{domain} или {service}:{entity}
• [ ] Consumer groups: одна группа на сервис-consumer
• [ ] MAXLEN или MINID: определить retention policy
• [ ] ID: использовать * (автогенерация) если не нужен кастомный порядок

Production

• [ ] Consumer как systemd-сервис с Restart=always
• [ ] XAUTOCLAIM для обработки «застрявших» записей
• [ ] Мониторинг: pending count, consumer lag, stream length
• [ ] AOF enabled: appendonly yes, appendfsync everysec
• [ ] Бэкапы: RDB snapshots + AOF
• [ ] XTRIM через cron: не допускать бесконтрольного роста streams
• [ ] Graceful shutdown: обработать SIGTERM, XACK перед выходом