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

Redis Is More Than a Cache

Most developers use Redis for two things: caching and queues (via Sidekiq/BullMQ). But Redis has long outgrown the role of a "fast key-value store." Pub/Sub for real-time notifications, Sorted Sets for leaderboards, HyperLogLog for unique counters, Streams for event-driven architecture — all built in and running at in-memory speed.

Redis Streams is the most underrated data structure. Introduced in Redis 5.0, it became truly mature in 7.x. It's essentially an append-only log (like Kafka), but with the Redis API and without needing to deploy a 6-node cluster with ZooKeeper.

In my projects, Streams replaced three things: custom Pub/Sub for real-time notifications, polling-based synchronization between services, and a separate queue for event sourcing. One Redis, one data structure — three problems solved.

What Are Redis Streams

The Concept

A Stream is an append-only log with an ID for each entry. Entries consist of key-value pairs. They're ordered by time and never modified after insertion.

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

ID format: <timestamp_ms>-<sequence>. Auto-generated (*) or set manually. Timestamp guarantees chronological order, sequence ensures uniqueness within a millisecond.

Basic Operations

# Add entry to stream
XADD orders * action created order_id 1001 amount 99.99

# Read all entries
XRANGE orders - +

# Read last 10
XREVRANGE orders + - COUNT 10

# Stream length
XLEN orders

Stream vs Pub/Sub vs List

Stream is the only structure combining persistence, re-reading, consumer groups, and acknowledgment.

Consumer Groups — Parallel Processing

Consumer Group distributes entries among multiple consumers. Each entry goes to exactly one consumer in the group (unlike Pub/Sub where everyone gets everything).

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

Two groups read one stream independently. Within a group, entries are distributed among consumers.

Create and Read

# Create consumer group
XGROUP CREATE orders payment-service $ MKSTREAM
XGROUP CREATE orders notification-service 0 MKSTREAM

# Read from group (blocking)
XREADGROUP GROUP payment-service payment-worker-1 COUNT 10 BLOCK 5000 STREAMS orders >

# Acknowledge processing
XACK orders payment-service 1716000000000-0

# View pending (unacknowledged) entries
XPENDING orders payment-service - + 10

# Claim stuck entries from a crashed consumer
XAUTOCLAIM orders payment-service new-worker-1 60000 0-0 COUNT 10

Practical Patterns

Event Bus Between Microservices

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

Activity Feed / Timeline

# Add activity
XADD user:123:feed * type comment post_id 456 text "Great article!"
XADD user:123:feed * type like post_id 789

# Last 20 feed items
XREVRANGE user:123:feed + - COUNT 20

# Limit feed size
XTRIM user:123:feed MAXLEN ~ 1000

Real-Time Analytics

class PageviewAggregator
  STREAM = "pageviews"
  GROUP = "aggregator"
  CONSUMER = "agg-#{Process.pid}"

  def run
    ensure_group_exists
    loop do
      entries = redis.xreadgroup(GROUP, CONSUMER, STREAM, ">",
                                 count: 100, 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
end

Audit Log

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

# History for last hour
XRANGE audit:company:42 1716000000000 +

Stream as audit log is cheaper than a PostgreSQL table for write-heavy scenarios. Can asynchronously transfer to PostgreSQL for long-term storage.

Integration with Rails

Consumer as Background Job

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

Integration with Node.js

import Redis from 'ioredis';
const redis = new Redis(process.env.REDIS_URL);

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) {
        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);
      }
    }
  }
}

Memory Management and Retention

# Limit by count (approximate, faster)
XADD mystream MAXLEN ~ 10000 * key value

# Limit by time (delete entries older than cutoff)
XTRIM mystream MINID ~ 1716000000000

# Cron: trim entries older than 7 days
0 * * * * redis-cli XTRIM events:orders MINID ~ $(date -d '7 days ago' +%s%3N)

Redis Streams vs Kafka

When Redis Streams Is Enough

• Under 100K messages per second
• Loss of 1-2 seconds of data on crash is acceptable
• Don't need exactly-once semantics
• Redis already in infrastructure
• Simple event bus between 2-5 services

When You Need Kafka

• Over 100K messages per second
• Guaranteed persistence (financial transactions)
• Exactly-once processing
• Event sourcing with long-term storage
• 10+ consumers per topic

Monitoring

Prometheus + redis_exporter

Key metrics:

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


Alerts

- alert: StreamPendingHigh
  expr: redis_stream_group_pending > 1000
  for: 5m

- alert: StreamConsumerLag
  expr: redis_stream_length - redis_stream_group_last_delivered_id > 10000
  for: 10m

Final Checklist

Before Using

• [ ] Determine pattern: event bus, activity feed, rate limiting, audit log
• [ ] Estimate throughput: < 100K msg/sec → Redis Streams, > 100K → Kafka
• [ ] Assess durability requirements: is loss of 1-2 seconds acceptable?
• [ ] Redis already in infrastructure? If not, evaluate whether to add it

Design

• [ ] Stream naming: events:{domain} or {service}:{entity}
• [ ] Consumer groups: one group per consumer service
• [ ] MAXLEN or MINID: define retention policy
• [ ] ID: use * (auto-generation) unless custom ordering needed

Production

• [ ] Consumer as systemd service with Restart=always
• [ ] XAUTOCLAIM for processing stuck entries
• [ ] Monitoring: pending count, consumer lag, stream length
• [ ] AOF enabled: appendonly yes, appendfsync everysec
• [ ] Backups: RDB snapshots + AOF
• [ ] XTRIM via cron: prevent uncontrolled stream growth
• [ ] Graceful shutdown: handle SIGTERM, XACK before exit