Redis Streams

Backend Fundamentals · Messaging & Event-Driven Architecture · 10 min read

1. What problem does this solve?

Motivation
Modern back‑ends often need to decouple producers (e.g., a web API that receives user actions) from consumers (e.g., background workers that send emails, update analytics, or trigger downstream services). The key requirements are:

• Reliability – a message should survive crashes and be processed at least once.
• Ordering – many workflows depend on the order of events (e.g., order status updates).
• Scalability – many producers and consumers should run concurrently without bottlenecks.
• Back‑pressure handling – producers should not be blocked by slow consumers.

Problem
Traditional in‑memory queues or simple Redis lists (LPUSH/RPOP) lack built‑in consumer groups, pending‑message tracking, or automatic rebalancing. They also provide only at‑most‑once semantics and no visibility into which consumer processed what.

Why it matters
If you cannot guarantee that every event is processed, you risk data loss, inconsistent state, or duplicate work. Without ordering guarantees, downstream services may see events out of sequence, breaking invariants.

Consequences of not using a proper stream

• Duplicate work – a consumer may re‑process a message if it crashes before acknowledging.
• Lost messages – if a consumer never acknowledges and the stream grows indefinitely, you may run out of memory.
• Hard to scale – adding more workers requires manual coordination or custom logic.

Real‑world usage

• Order processing pipelines (e.g., e‑commerce order status updates).
• Event sourcing for domain models.
• Background job queues (image processing, email sending).
• Log aggregation and analytics pipelines.

2. Mental model

Analogy

Imagine a public mailbox that accepts envelopes (messages). Each envelope has a unique serial number (ID). Multiple mail carriers (consumers) can pick up envelopes from the mailbox. When a carrier takes an envelope, they write a note in a shared ledger (consumer group state) that they are handling it. If a carrier drops the envelope (fails before finishing), the ledger still shows it as pending, and another carrier can claim it later.

Engineering model

• Stream = ordered log of entries (envelopes).
• Consumer group = shared ledger of which consumer is processing which entry.
• Consumer = a worker that reads from the stream and acknowledges.
• Pending entries = entries that have been delivered to a consumer but not yet acknowledged.

3. Deep explanation

Internals

• Redis stores a stream as a linked list of entries in memory, each entry having a unique ID (<timestamp>-<sequence>).
• Each entry is a hash of field/value pairs.
• Consumer groups maintain a cursor (offset) per consumer, stored in a separate hash.
• Pending entries are tracked in a ZSET keyed by consumer group, mapping entry IDs to the consumer that owns them.

Important terminology

Trade‑offs

Failure modes

• Consumer crash – pending entries remain unacknowledged; must be claimed later.
• Redis crash – data may be lost if AOF/RDB not configured; replication mitigates.
• Network partition – consumers may read stale data; Redis Cluster can help.
• Stream growth – if not trimmed, memory usage can explode.

Limitations

• No built‑in TTL for entries (must trim manually).
• No exactly‑once semantics.
• Limited to a single Redis instance per stream unless using Redis Cluster.
• Consumer group rebalancing is manual; no automatic partition assignment.

Where it fits

• Message broker – acts as a lightweight broker for event streams.
• Task queue – workers consume tasks from a stream.
• Event sourcing – domain events are appended to a stream.
• Change data capture – database changes can be streamed into Redis.

Interaction with related concepts

• Pub/Sub – Redis Streams provide persistence and consumer groups; Pub/Sub is fire‑and‑forget.
• Redis Lists – simpler but lack consumer groups and pending entry tracking.
• Kafka – similar semantics but more feature‑rich (exactly‑once, compaction, multi‑topic partitions).
• Database triggers – can push events into a stream for downstream processing.

4. How it works

1. Producer calls XADD streamKey * field1 value1 field2 value2 … to append an entry.
2. Consumer group myGroup is created with XGROUP CREATE streamKey myGroup $.
3. Consumer worker-1 reads new entries:
   XREADGROUP GROUP myGroup worker-1 COUNT 10 BLOCK 0 STREAMS streamKey >.
   The > means “new entries only”.
4. Redis delivers the entries to worker-1 and records them as pending for that consumer.
5. worker-1 processes each entry and acknowledges with XACK streamKey myGroup entryID.
6. If worker-1 crashes before acknowledging, the entry remains pending.
7. Another consumer can claim it after a timeout using XCLAIM streamKey myGroup newConsumer idleTime entryID.
8. Periodically, the stream can be trimmed with XTRIM streamKey MAXLEN 100000 to free memory.

sequenceDiagram
    participant Prod as Producer
    participant Redis
    participant C1 as Consumer1
    participant C2 as Consumer2

    Prod->>Redis: XADD streamKey * field1 val1
    Redis->>C1: Deliver entry (pending for C1)
    C1->>Redis: XACK streamKey myGroup entryID
    Note over C1: Entry processed

    Prod->>Redis: XADD streamKey * field2 val2
    Redis->>C1: Deliver entry (pending for C1)
    C1->>Redis: Crash (no ACK)
    Redis->>C2: XCLAIM after idle timeout
    C2->>Redis: XACK streamKey myGroup entryID

5. Practical implementation

Setup

# Install dependencies
npm install redis

Producer (orders.ts)

import { createClient } from 'redis';

const client = createClient();
await client.connect();

export async function produceOrder(order: { id: string; status: string }) {
  // Use auto‑generated ID (*)
  await client.xAdd('orders', '*', {
    id: order.id,
    status: order.status,
  });
}

Consumer group creation (init.ts)

import { createClient } from 'redis';

const client = createClient();
await client.connect();

// Create consumer group if it doesn't exist
try {
  await client.xGroupCreate('orders', 'order_workers', '
  

, { mkstream: true });
} catch (e) {
  if (!e.message.includes('BUSYGROUP')) throw e;
}

Worker (worker.ts)

import { createClient } from 'redis';

const client = createClient();
await client.connect();

const CONSUMER_ID = `worker-${process.pid}`;

async function processEntry(entry: { id: string; fields: Record<string, string> }) {
  const { id, fields } = entry;
  console.log(`Processing order ${fields.id} with status ${fields.status}`);
  // Simulate work
  await new Promise((r) => setTimeout(r, 500));

  // Acknowledge
  await client.xAck('orders', 'order_workers', id);
}

async function run() {
  while (true) {
    // Read up to 10 new entries, block if none
    const streams = await client.xReadGroup(
      'order_workers',
      CONSUMER_ID,
      { key: 'orders', id: '>' },
      { count: 10, block: 0 }
    );

    if (!streams) continue;

    for (const stream of streams) {
      for (const [id, fields] of stream.messages) {
        try {
          await processEntry({ id, fields: fields as Record<string, string> });
        } catch (e) {
          console.error('Processing failed', e);
          // Optionally leave the entry pending for retry
        }
      }
    }
  }
}

run().catch(console.error);

Handling pending entries (recovery.ts)

import { createClient } from 'redis';

const client = createClient();
await client.connect();

async function claimStaleEntries() {
  // Claim entries idle > 60s
  const entries = await client.xClaim(
    'orders',
    'order_workers',
    'recovery-worker',
    60000,
    ['*'] // all pending entries
  );

  for (const [id, fields] of entries) {
    console.log(`Claimed stale entry ${id}`);
    // Process and ack
    await client.xAck('orders', 'order_workers', id);
  }
}

claimStaleEntries().catch(console.error);

6. Production considerations

7. Common mistakes

8. Senior engineer perspective

Design decisions at scale

• Stream partitioning: Use multiple streams (orders:shard-1, orders:shard-2) to avoid a single hot key.
• Consumer group strategy: One group per logical service; avoid cross‑service groups unless needed.
• Retention policy: Set MAXLEN or use XTRIM with ~ for approximate trimming to keep memory bounded.
• Cluster vs Sentinel: Prefer Redis Cluster for horizontal scaling; Sentinel for simpler HA.
• Monitoring: Expose redis_stream_pending and redis_stream_consumers to Prometheus; alert on high pending counts.
• Cost: Each entry consumes ~100 bytes; 1 M entries ≈ 100 MB. Scale memory accordingly.

Bottlenecks

• Memory: Streams are in‑memory; large streams can saturate RAM.
• CPU: XCLAIM and XACK are O(1), but large pending sets can slow down.
• Network: High throughput streams can saturate network if workers are remote.

Failure scenarios

• Consumer crash → pending entries remain; must be claimed.
• Redis node failure → data loss if not replicated; use AOF or RDB.
• Network partition → consumers may read stale data; use Redis Cluster with quorum.

When NOT to use

• Exactly‑once semantics required → consider Kafka or a transactional database.
• Cross‑region replication needed → Redis Cluster may not span regions easily.
• Complex routing (e.g., topic‑based) → use a full‑blown broker.

9. Interview questions

10. Hands‑on task

Build an order‑processing pipeline

1. Create a Node.js project with redis client.
2. Implement a producer that reads orders from a JSON file and pushes them to a stream orders.
3. Create a consumer group order_workers.
4. Spin up 3 worker processes that read from the stream, simulate processing (e.g., sleep 200 ms), and acknowledge.
5. Add logic to claim pending entries idle > 30 s.
6. Periodically trim the stream to keep at most 500 k entries.
7. Expose a simple HTTP endpoint /stats that returns the number of pending entries and last processed order ID.
8. Run the system, observe that workers process orders, pending entries stay low, and the /stats endpoint reports correct values.

11. Quick revision

• Redis Streams are append‑only logs with unique IDs (<timestamp>-<sequence>).
• XADD appends; XREADGROUP reads; XACK acknowledges.
• Consumer groups share a cursor; pending entries track unacknowledged messages.
• XCLAIM reclaims stale pending entries.
• Streams are in‑memory; trim with XTRIM or MAXLEN.
• Use XGROUP CREATE with mkstream to create a stream if it doesn’t exist.
• For idempotent processing, design consumers to handle duplicates.
• Monitor pending counts (XPENDING) and stream length (XLEN).
• Use Redis Cluster or Sentinel for HA.
• Avoid XREAD without consumer groups; it lacks pending tracking.
• Exactly‑once semantics are not guaranteed; use idempotency or external deduplication.
• Trimming is essential to control memory usage.

12. References

• Redis Streams – Official Documentation
• Redis Streams – Commands
• node-redis v4 API
• ioredis – Streams Example
• Redis Cluster – Streams
• Redis Sentinel – High Availability
• Redis Streams – Best Practices