Redis Streams: Event Bus, Rate Limiting, Audit Log ohne Kafka

Redis ist mehr als ein Cache

Die meisten Entwickler nutzen Redis für zwei Dinge: Caching und Queues (über Sidekiq/BullMQ). Aber Redis ist längst über die Rolle eines „schnellen Key-Value-Speichers" hinausgewachsen. Pub/Sub für Echtzeit-Benachrichtigungen, Sorted Sets für Bestenlisten, HyperLogLog für eindeutige Zähler, Streams für Event-Driven-Architektur — alles eingebaut und mit In-Memory-Geschwindigkeit.

Redis Streams ist die am meisten unterschätzte Datenstruktur. In Redis 5.0 eingeführt, wurde sie in 7.x wirklich ausgereift. Im Wesentlichen ein Append-Only-Log (wie Kafka), aber mit der Redis-API und ohne einen 6-Knoten-Cluster mit ZooKeeper aufsetzen zu müssen.

In meinen Projekten hat Streams drei Dinge ersetzt: eigenes Pub/Sub für Echtzeit-Benachrichtigungen, Polling-basierte Synchronisation zwischen Services und eine separate Queue für Event Sourcing. Ein Redis, eine Datenstruktur — drei Probleme gelöst.

Was sind Redis Streams

Das Konzept

Ein Stream ist ein Append-Only-Log mit einer ID pro Eintrag. Einträge bestehen aus Schlüssel-Wert-Paaren, sind nach Zeit geordnet und werden nach dem Hinzufügen nie geändert.

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       │
└──────────────────┘────────────────────────────────────┘

Grundlegende Operationen

# Eintrag zum Stream hinzufügen
XADD orders * action created order_id 1001 amount 99.99

# Alle Einträge lesen
XRANGE orders - +

# Letzte 10
XREVRANGE orders + - COUNT 10

# Stream-Länge
XLEN orders

Stream vs Pub/Sub vs List

Consumer Groups — Parallele Verarbeitung

Consumer Group verteilt Einträge auf mehrere Consumer. Jeder Eintrag geht an genau einen Consumer in der Gruppe.

Stream: orders
    ├── Consumer Group: "payment-service"
    │       ├── Consumer: payment-1  ← bekommt Eintrag A
    │       └── Consumer: payment-2  ← bekommt Eintrag B
    │
    └── Consumer Group: "notification-service"
            ├── Consumer: notif-1    ← bekommt Eintrag A
            └── Consumer: notif-2    ← bekommt Eintrag B

# Consumer Group erstellen
XGROUP CREATE orders payment-service $ MKSTREAM

# Aus Gruppe lesen (blockierend)
XREADGROUP GROUP payment-service worker-1 COUNT 10 BLOCK 5000 STREAMS orders >

# Verarbeitung bestätigen
XACK orders payment-service 1716000000000-0

# Feststeckende Einträge übernehmen
XAUTOCLAIM orders payment-service new-worker 60000 0-0 COUNT 10

Praktische Muster

Event Bus zwischen Microservices

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
end

Rate Limiting

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)
    if r.xlen(stream_key) >= limit:
        return True
    r.xadd(stream_key, {"ts": now}, maxlen=limit * 2)
    return False

Activity Feed

XADD user:123:feed * type comment post_id 456 text "Toller Artikel!"
XREVRANGE user:123:feed + - COUNT 20
XTRIM user:123:feed MAXLEN ~ 1000

Audit Log

XADD audit:company:42 * \
  user_id 5 action update entity Employee entity_id 123 \
  changes '{"salary": [50000, 55000]}'

Stream als Audit-Log ist günstiger als eine PostgreSQL-Tabelle für schreibintensive Szenarien.

Integration mit Rails

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)
        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
    end
  end
end

Speicherverwaltung und Retention

# Nach Anzahl begrenzen
XADD mystream MAXLEN ~ 10000 * key value

# Nach Zeit begrenzen
XTRIM mystream MINID ~ 1716000000000

# Cron: Einträge älter als 7 Tage trimmen
0 * * * * redis-cli XTRIM events:orders MINID ~ $(date -d '7 days ago' +%s%3N)

Redis Streams vs Kafka

Wann Redis Streams ausreicht

• Unter 100K Nachrichten pro Sekunde
• Verlust von 1-2 Sekunden bei Absturz akzeptabel
• Redis bereits in der Infrastruktur
• Einfacher Event Bus zwischen 2-5 Services

Wann Kafka nötig ist

• Über 100K Nachrichten pro Sekunde
• Garantierte Persistenz (Finanztransaktionen)
• Exactly-once-Verarbeitung
• Event Sourcing mit Langzeitspeicherung

Monitoring

Prometheus + redis_exporter

Wichtige Metriken:

redis_stream_length{stream="events:orders"}
redis_stream_group_pending{stream="events:orders",group="payment-service"}


Abschließende Checkliste

Vor der Nutzung

• [ ] Muster bestimmen: Event Bus, Activity Feed, Rate Limiting, Audit Log
• [ ] Durchsatz abschätzen: < 100K msg/sec → Redis Streams, > 100K → Kafka
• [ ] Haltbarkeitsanforderungen bewerten: Verlust von 1-2 Sekunden akzeptabel?

Design

• [ ] Stream-Benennung: events:{domain} oder {service}:{entity}
• [ ] Consumer Groups: eine Gruppe pro Consumer-Service
• [ ] MAXLEN oder MINID: Retention-Policy definieren

Produktion

• [ ] Consumer als systemd-Service mit Restart=always
• [ ] XAUTOCLAIM für feststeckende Einträge
• [ ] Monitoring: Pending Count, Consumer Lag, Stream-Länge
• [ ] AOF aktiviert: appendonly yes, appendfsync everysec
• [ ] Backups: RDB-Snapshots + AOF
• [ ] XTRIM via Cron: unkontrolliertes Wachstum verhindern
• [ ] Graceful Shutdown: SIGTERM behandeln, XACK vor Beendigung