Redis chevron_right Redis Streams - Advanced Data Processing and Event Streaming

Redis Streams - Advanced Data Processing and Event Streaming

calendar_today schedule 28 min read

Learning Objectives: By the end of this lesson, you will be able to:

  • Understand Redis Streams and how they differ from other Redis data structures
  • Implement event streaming and message processing with Redis Streams
  • Build consumer groups for distributed message processing
  • Handle stream trimming, acknowledgments, and pending messages
  • Create production-ready stream processing applications in Golang

1. Welcome and Introduction (5 minutes)

Icebreaker Activity

Have you ever wondered how large-scale systems like Twitter, Facebook, or e-commerce platforms process millions of events per second—clicks, purchases, notifications—without losing a single message? How do they ensure that multiple services can consume the same events without interfering with each other?

The answer lies in event streaming systems. Today, we’ll explore Redis Streams, a powerful data structure that brings enterprise-grade event streaming capabilities to Redis, allowing you to build sophisticated real-time data pipelines with minimal complexity.

Lesson Overview

  • Duration: 90-120 minutes
  • Prerequisites:
    • Understanding of basic Redis data structures (Lists, Pub/Sub)
    • Familiarity with Golang and go-redis library
    • Basic knowledge of concurrent programming
    • Understanding of distributed systems concepts
  • Materials Needed:
    • Redis 5.0+ (Streams were introduced in Redis 5.0)
    • Go 1.19+
    • go-redis/redis/v9 library
    • Code editor and terminal access

2. Learning Objectives and Session Outline (5 minutes)

What We’ll Cover Today:

  1. Redis Streams Fundamentals - Understanding the stream data structure, its advantages, and use cases
  2. Basic Stream Operations - Adding entries, reading streams, and range queries
  3. Consumer Groups - Implementing reliable distributed message processing
  4. Advanced Features - Pending entries, claiming messages, and stream trimming
  5. Practical Implementation - Building a real-world event processing system in Golang
  6. Performance and Best Practices - Optimizing stream operations and handling edge cases

Success Criteria:

By the end of this session, you should be able to:

  • Create and manage Redis Streams for event processing
  • Implement consumer groups for parallel message processing
  • Handle failures and ensure message delivery guarantees
  • Build production-ready stream processing applications
  • Understand when to use Streams vs Pub/Sub vs Lists

3. Foundation Knowledge (15-20 minutes)

Core Concepts

What Are Redis Streams?

Redis Streams is an append-only log data structure that allows you to store and process a continuous flow of data entries. Think of it as a sophisticated, persistent message queue that combines the best features of Redis Lists and Pub/Sub while adding powerful new capabilities.

Real-World Analogy: Imagine a conveyor belt in a factory where items (messages) continuously arrive. Multiple workers (consumers) can process these items, and if a worker gets sick, another can pick up where they left off. The conveyor belt keeps a record of which items each worker has processed, ensuring nothing gets lost.

Why This Matters: Unlike Redis Pub/Sub (where messages are lost if no subscriber is listening) and Lists (which lack consumer group semantics), Streams provide:

  • Persistence: Messages remain in the stream until explicitly deleted
  • Consumer Groups: Multiple consumers can coordinate without conflicts
  • Message Acknowledgment: Track which messages have been processed
  • Automatic ID Generation: Each entry gets a unique, time-based ID
  • Range Queries: Read historical data efficiently
  • Blocking Reads: Wait for new messages without polling

Redis Streams vs Other Data Structures

Streams vs Pub/Sub:

  • Pub/Sub: Fire-and-forget, no persistence, real-time delivery
  • Streams: Persistent, replayable, supports consumer groups, guaranteed delivery

Streams vs Lists:

  • Lists: Simple FIFO queue, no consumer group support, manual coordination needed
  • Streams: Built-in consumer groups, automatic message IDs, rich querying capabilities

Streams vs Sorted Sets:

  • Sorted Sets: Random access by score, not optimized for streaming
  • Streams: Optimized for append-only operations, time-series data, and sequential processing

Stream Entry Structure

Each entry in a Redis Stream consists of:

  1. Stream ID: Automatically generated timestamp-based ID (e.g., 1609459200000-0)
    • Format: <millisecond-timestamp>-<sequence-number>
    • Ensures ordering and uniqueness
  2. Field-Value Pairs: Arbitrary key-value data (like a hash)

Consumer Groups Architecture

Consumer Groups enable multiple consumers to cooperate in processing stream entries:

  • Group: A named collection of consumers
  • Consumer: Individual worker within a group
  • Pending Entries List (PEL): Tracks messages delivered but not yet acknowledged
  • Last Delivered ID: Tracks the group’s progress through the stream

Key Terminology

  • Stream: An append-only log data structure that stores entries with unique IDs
  • Entry: A single message in the stream with an ID and field-value pairs
  • Consumer Group: A coordination mechanism allowing multiple consumers to process entries without duplication
  • Consumer: An individual worker within a consumer group that processes stream entries
  • Pending Entry: A message delivered to a consumer but not yet acknowledged
  • Claimed Entry: A pending message reassigned from an inactive consumer to an active one
  • Trimming: Removing old entries from a stream to manage memory usage
  • Blocking Read: Waiting for new entries to arrive rather than polling repeatedly

4. Detailed Explanation and Examples (25-30 minutes)

Understanding Redis Streams Architecture

Redis Streams are designed to handle high-throughput event streaming with minimal latency. The data structure is optimized for:

  1. Fast Appends: O(1) complexity for adding new entries
  2. Efficient Range Queries: Retrieve entries by ID range or time window
  3. Consumer Group Coordination: Built-in mechanisms for distributed processing
  4. Memory Efficiency: Compact storage with configurable retention policies

Stream ID Generation

Stream IDs follow the format <milliseconds>-<sequence>. Redis automatically generates IDs based on the server’s timestamp, ensuring:

  • Monotonic Ordering: IDs always increase
  • Time-Based Queries: Easy to query entries by time range
  • Uniqueness: Sequence number prevents collisions within the same millisecond

You can also specify custom IDs (must be greater than the last ID) for specialized use cases like importing historical data.

Consumer Group Semantics

Consumer groups solve the “competing consumers” problem in distributed systems. Without consumer groups, multiple workers would need complex coordination to avoid processing the same message twice. With consumer groups:

  1. Automatic Load Distribution: Each entry is delivered to exactly one consumer in the group
  2. Failure Recovery: If a consumer crashes, its pending messages can be reassigned
  3. Parallel Processing: Multiple groups can independently consume the same stream
  4. Progress Tracking: Each group maintains its own position in the stream

Practical Applications

Use Case 1: Event Sourcing Store all state changes as immutable events. Applications can replay events to rebuild state or create projections.

Use Case 2: Activity Feeds Track user actions (likes, comments, posts) in real-time. Multiple services can consume the feed independently.

Use Case 3: Log Aggregation Collect logs from distributed services. Different teams can analyze logs without interfering with each other.

Use Case 4: IoT Data Ingestion Process sensor data from thousands of devices. Scale processing by adding consumers to groups.

Use Case 5: Distributed Task Queue Distribute work across multiple workers with guaranteed delivery and automatic retry on failure.

Common Misconceptions

  1. Misconception: “Redis Streams are just like Kafka, so I should always use Kafka instead for serious workloads.” Reality: While Kafka excels at massive scale and complex multi-datacenter deployments, Redis Streams offers lower latency, simpler operations, and seamless integration with Redis’s other features. For many applications (especially those already using Redis), Streams provides sufficient throughput (hundreds of thousands of messages per second) with far less operational complexity.

  2. Misconception: “I should always acknowledge messages immediately after receiving them.” Reality: Acknowledge messages only after successfully processing them. Premature acknowledgment can lead to data loss if processing fails. The pending entries list exists precisely to track unacknowledged messages.

  3. Misconception: “Consumer groups are just for load balancing.” Reality: Consumer groups provide multiple benefits: load distribution, fault tolerance, independent processing by different applications, and progress tracking. Even a single-consumer group is useful for tracking processing progress and ensuring at-least-once delivery.

  4. Misconception: “Streams use too much memory, so I should trim aggressively.” Reality: Redis Streams are memory-efficient (typically 10-50 bytes per entry plus data). Balance retention with business needs. Keep enough history for failure recovery and debugging while trimming old entries that are no longer needed.

5. Live Coding Demonstration (20-25 minutes)

Example 1: Basic Stream Operations

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    
    // Connect to Redis
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    // Stream name
    streamKey := "events:user-activity"

    // 1. Add entries to the stream
    // XADD command with auto-generated ID (*)
    id1, err := rdb.XAdd(ctx, &redis.XAddArgs{
        Stream: streamKey,
        Values: map[string]interface{}{
            "user_id": "user123",
            "action":  "login",
            "ip":      "192.168.1.100",
        },
    }).Result()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Added entry with ID: %s\n", id1)

    // Add more entries
    id2, err := rdb.XAdd(ctx, &redis.XAddArgs{
        Stream: streamKey,
        Values: map[string]interface{}{
            "user_id": "user456",
            "action":  "purchase",
            "amount":  "99.99",
            "item":    "laptop",
        },
    }).Result()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Added entry with ID: %s\n", id2)

    // 2. Read entries from the stream
    // XREAD - read from beginning (0-0 means start)
    streams, err := rdb.XRead(ctx, &redis.XReadArgs{
        Streams: []string{streamKey, "0"}, // stream name, start ID
        Count:   10,                        // max entries to read
        Block:   0,                         // 0 = don't block, return immediately
    }).Result()
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("\n--- Reading stream entries ---")
    for _, stream := range streams {
        for _, message := range stream.Messages {
            fmt.Printf("ID: %s\n", message.ID)
            fmt.Printf("  Data: %v\n", message.Values)
        }
    }

    // 3. Read specific range
    // XRANGE - get entries between IDs
    messages, err := rdb.XRange(ctx, streamKey, "-", "+").Result()
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("\n--- Range query (all entries) ---")
    for _, msg := range messages {
        fmt.Printf("ID: %s, Values: %v\n", msg.ID, msg.Values)
    }

    // 4. Get stream length
    length, err := rdb.XLen(ctx, streamKey).Result()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("\nStream length: %d entries\n", length)
}

Explanation:

  • Line 23: XAdd appends a new entry to the stream with auto-generated ID
  • Line 24: Using * as ID tells Redis to generate a timestamp-based ID automatically
  • Line 25-29: Field-value pairs stored in the entry (like a hash)
  • Line 47: XRead retrieves entries from the stream, starting from ID “0” (beginning)
  • Line 48: Count: 10 limits the number of entries returned
  • Line 49: Block: 0 returns immediately; positive values wait for new entries
  • Line 64: XRange queries entries between two IDs; “-” and “+” mean start and end
  • Overall: This demonstrates the fundamental operations for adding and reading stream entries

Example 2: Consumer Groups - Producer and Consumer

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

// Producer: Generates events
func producer(ctx context.Context, rdb *redis.Client, streamKey string) {
    fmt.Println("Producer started...")
    
    for i := 0; i < 20; i++ {
        id, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamKey,
            Values: map[string]interface{}{
                "event_id": i,
                "message":  fmt.Sprintf("Event number %d", i),
                "timestamp": time.Now().Unix(),
            },
        }).Result()
        if err != nil {
            log.Printf("Producer error: %v", err)
            continue
        }
        fmt.Printf("Produced: %s (event %d)\n", id, i)
        time.Sleep(500 * time.Millisecond) // Simulate event generation
    }
    fmt.Println("Producer finished")
}

// Consumer: Processes events using consumer group
func consumer(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string) {
    fmt.Printf("Consumer %s started...\n", consumerName)

    // Create consumer group if it doesn't exist
    // XGROUP CREATE creates a group starting from the end ($) or beginning (0)
    err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0").Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        log.Printf("Error creating group: %v", err)
        return
    }

    for {
        // XREADGROUP - read as part of a consumer group
        streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    groupName,
            Consumer: consumerName,
            Streams:  []string{streamKey, ">"}, // ">" means undelivered messages
            Count:    1,                         // Process one at a time
            Block:    2 * time.Second,           // Wait up to 2 seconds for new messages
        }).Result()

        if err != nil {
            if err == redis.Nil {
                // No new messages, timeout reached
                fmt.Printf("[%s] No new messages, waiting...\n", consumerName)
                continue
            }
            log.Printf("Consumer %s error: %v", consumerName, err)
            break
        }

        // Process messages
        for _, stream := range streams {
            for _, message := range stream.Messages {
                fmt.Printf("[%s] Processing: ID=%s, Data=%v\n", 
                    consumerName, message.ID, message.Values)
                
                // Simulate processing work
                time.Sleep(1 * time.Second)

                // XACK - acknowledge message after successful processing
                err := rdb.XAck(ctx, streamKey, groupName, message.ID).Err()
                if err != nil {
                    log.Printf("[%s] Failed to ack %s: %v", consumerName, message.ID, err)
                } else {
                    fmt.Printf("[%s] Acknowledged: %s\n", consumerName, message.ID)
                }
            }
        }
    }
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "events:processing"
    groupName := "processing-group"

    // Clean up from previous runs
    rdb.Del(ctx, streamKey)

    // Start producer in goroutine
    go producer(ctx, rdb, streamKey)

    // Start multiple consumers in the same group
    go consumer(ctx, rdb, streamKey, groupName, "consumer-1")
    go consumer(ctx, rdb, streamKey, groupName, "consumer-2")

    // Let them run for a while
    time.Sleep(15 * time.Second)
    fmt.Println("\nMain: Shutting down...")
}

Explanation:

  • Line 41: XGroupCreateMkStream creates a consumer group and the stream if it doesn’t exist
  • Line 41: Starting position “0” means process from the beginning; “$” would mean only new messages
  • Line 49: XReadGroup reads messages as part of a consumer group
  • Line 50: Each consumer has a unique name within the group
  • Line 51: “>” as the ID means “give me messages never delivered to any consumer in this group”
  • Line 53: Block makes the consumer wait for new messages instead of polling
  • Line 77: XAck acknowledges successful processing, removing from pending list
  • Overall: This demonstrates how multiple consumers in a group automatically distribute work without processing duplicates

Example 3: Handling Pending Messages and Failures

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

// RobustConsumer handles pending messages and failures
func robustConsumer(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string) {
    fmt.Printf("[%s] Starting robust consumer...\n", consumerName)

    for {
        // First, check for pending messages (messages delivered but not acknowledged)
        pendingMessages, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    groupName,
            Consumer: consumerName,
            Streams:  []string{streamKey, "0"}, // "0" means check pending messages
            Count:    10,
            Block:    0, // Don't block for pending messages
        }).Result()

        if err != nil && err != redis.Nil {
            log.Printf("[%s] Error reading pending: %v", consumerName, err)
        }

        // Process pending messages first (recovery from previous failures)
        if len(pendingMessages) > 0 {
            fmt.Printf("[%s] Found %d pending messages\n", consumerName, len(pendingMessages[0].Messages))
            for _, stream := range pendingMessages {
                for _, msg := range stream.Messages {
                    if processMessage(consumerName, msg) {
                        rdb.XAck(ctx, streamKey, groupName, msg.ID)
                        fmt.Printf("[%s] Recovered and acked: %s\n", consumerName, msg.ID)
                    }
                }
            }
        }

        // Now read new messages
        streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    groupName,
            Consumer: consumerName,
            Streams:  []string{streamKey, ">"},
            Count:    1,
            Block:    2 * time.Second,
        }).Result()

        if err != nil {
            if err == redis.Nil {
                continue
            }
            log.Printf("[%s] Error: %v", consumerName, err)
            break
        }

        for _, stream := range streams {
            for _, msg := range stream.Messages {
                if processMessage(consumerName, msg) {
                    rdb.XAck(ctx, streamKey, groupName, msg.ID)
                } else {
                    fmt.Printf("[%s] Failed to process %s (left in pending)\n", consumerName, msg.ID)
                }
            }
        }
    }
}

// processMessage simulates message processing with potential failures
func processMessage(consumerName string, msg redis.XMessage) bool {
    fmt.Printf("[%s] Processing: ID=%s, Data=%v\n", consumerName, msg.ID, msg.Values)
    
    // Simulate processing
    time.Sleep(500 * time.Millisecond)
    
    // Simulate occasional failures (10% failure rate)
    if time.Now().Unix()%10 == 0 {
        fmt.Printf("[%s] ❌ Processing failed for %s\n", consumerName, msg.ID)
        return false
    }
    
    fmt.Printf("[%s] ✓ Successfully processed %s\n", consumerName, msg.ID)
    return true
}

// ClaimStaleMessages reassigns messages from dead consumers
func claimStaleMessages(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string) {
    // XPENDING returns information about pending messages
    pending, err := rdb.XPendingExt(ctx, &redis.XPendingExtArgs{
        Stream: streamKey,
        Group:  groupName,
        Start:  "-",
        End:    "+",
        Count:  100,
    }).Result()

    if err != nil {
        log.Printf("Error checking pending: %v", err)
        return
    }

    // Check for messages pending longer than 10 seconds
    staleThreshold := 10 * time.Second
    var staleIDs []string

    for _, msg := range pending {
        if msg.Idle > staleThreshold {
            staleIDs = append(staleIDs, msg.ID)
            fmt.Printf("Found stale message: %s (idle for %v, consumer: %s)\n", 
                msg.ID, msg.Idle, msg.Consumer)
        }
    }

    // XCLAIM - claim stale messages from other consumers
    if len(staleIDs) > 0 {
        claimed, err := rdb.XClaim(ctx, &redis.XClaimArgs{
            Stream:   streamKey,
            Group:    groupName,
            Consumer: consumerName,
            MinIdle:  staleThreshold,
            Messages: staleIDs,
        }).Result()

        if err != nil {
            log.Printf("Error claiming messages: %v", err)
            return
        }

        fmt.Printf("[%s] Claimed %d stale messages\n", consumerName, len(claimed))
        
        // Process claimed messages
        for _, msg := range claimed {
            if processMessage(consumerName, msg) {
                rdb.XAck(ctx, streamKey, groupName, msg.ID)
            }
        }
    }
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "events:robust"
    groupName := "robust-group"

    // Clean up
    rdb.Del(ctx, streamKey)

    // Create group
    rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0")

    // Add some test messages
    for i := 0; i < 10; i++ {
        rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamKey,
            Values: map[string]interface{}{
                "task_id": i,
                "data":    fmt.Sprintf("Task %d", i),
            },
        })
    }

    // Start robust consumer
    go robustConsumer(ctx, rdb, streamKey, groupName, "consumer-1")

    // Periodically check for stale messages
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            claimStaleMessages(ctx, rdb, streamKey, groupName, "consumer-1")
        }
    }()

    time.Sleep(30 * time.Second)
    fmt.Println("\nShutting down...")
}

Key Points:

  • Pending Message Recovery: Always check for pending messages on startup to handle previous failures
  • XPendingExt: Provides detailed information about unacknowledged messages
  • XClaim: Allows taking over messages from dead or slow consumers
  • Idempotency: Ensure message processing is idempotent since messages might be processed multiple times

Debugging Demonstration

Common errors and how to fix them:

// Common mistake: Forgetting to create consumer group
streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "nonexistent-group",
    Consumer: "consumer-1",
    Streams:  []string{"mystream", ">"},
}).Result()
// Error: NOGROUP No such key 'mystream' or consumer group 'nonexistent-group'

// Corrected version: Always create group first
err := rdb.XGroupCreateMkStream(ctx, "mystream", "nonexistent-group", "0").Err()
if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
    log.Fatal(err)
}

// Now safe to read
streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "nonexistent-group",
    Consumer: "consumer-1",
    Streams:  []string{"mystream", ">"},
}).Result()

6. Hands-On Practice (30-40 minutes)

Exercise 1: Basic Stream Producer-Consumer (10 minutes)

Objective: Create a simple order processing system using Redis Streams

Instructions:

  1. Create a producer that generates order events with fields: order_id, customer_id, amount, timestamp
  2. Create a consumer that reads these orders and prints them
  3. Add 10 orders and verify all are processed
  4. Check the stream length before and after processing

Starter Code:

package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "orders:new"

    // TODO: Implement producer
    // Add 10 order events with order_id, customer_id, amount

    // TODO: Implement consumer
    // Read all orders from the stream and print them

    // TODO: Check stream length
}

Solution:

package main

import (
    "context"
    "fmt"
    "log"
    "math/rand"
    "time"

    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "orders:new"

    // Clean up from previous runs
    rdb.Del(ctx, streamKey)

    // Producer: Add 10 order events
    fmt.Println("=== Producing Orders ===")
    for i := 1; i <= 10; i++ {
        id, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamKey,
            Values: map[string]interface{}{
                "order_id":    fmt.Sprintf("ORD-%d", 1000+i),
                "customer_id": fmt.Sprintf("CUST-%d", rand.Intn(100)),
                "amount":      fmt.Sprintf("%.2f", rand.Float64()*500),
                "timestamp":   time.Now().Unix(),
            },
        }).Result()
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("Produced order %d with ID: %s\n", i, id)
    }

    // Check stream length
    length, err := rdb.XLen(ctx, streamKey).Result()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("\nStream length: %d orders\n\n", length)

    // Consumer: Read all orders
    fmt.Println("=== Consuming Orders ===")
    messages, err := rdb.XRead(ctx, &redis.XReadArgs{
        Streams: []string{streamKey, "0"},
        Count:   100,
        Block:   0,
    }).Result()
    if err != nil {
        log.Fatal(err)
    }

    for _, stream := range messages {
        for i, msg := range stream.Messages {
            fmt.Printf("Order %d:\n", i+1)
            fmt.Printf("  ID: %s\n", msg.ID)
            fmt.Printf("  Order ID: %s\n", msg.Values["order_id"])
            fmt.Printf("  Customer: %s\n", msg.Values["customer_id"])
            fmt.Printf("  Amount: $%s\n", msg.Values["amount"])
            fmt.Printf("  Timestamp: %s\n\n", msg.Values["timestamp"])
        }
    }

    fmt.Printf("Processed %d orders\n", len(messages[0].Messages))
}

Exercise 2: Consumer Group with Multiple Consumers (15 minutes)

Objective: Implement a distributed task processing system with consumer groups

Requirements:

  • Create a stream for tasks
  • Create a consumer group called “task-workers”
  • Start 3 consumers in the same group
  • Producer should add 20 tasks over 10 seconds
  • Verify that tasks are distributed among consumers
  • Each consumer should acknowledge processed tasks

Hints:

  • Use goroutines for concurrent consumers
  • Add logging to track which consumer processes which task
  • Use time.Sleep to simulate task processing time
  • Remember to create the consumer group before starting consumers

Starter Code:

package main

import (
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

func producer(ctx context.Context, rdb *redis.Client, streamKey string, wg *sync.WaitGroup) {
    defer wg.Done()
    // TODO: Generate 20 tasks
}

func consumer(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string, wg *sync.WaitGroup) {
    defer wg.Done()
    // TODO: Implement consumer logic with XReadGroup and XAck
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "tasks:processing"
    groupName := "task-workers"

    // TODO: Create consumer group
    
    var wg sync.WaitGroup

    // TODO: Start producer and consumers
    
    wg.Wait()
}

Solution:

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

func producer(ctx context.Context, rdb *redis.Client, streamKey string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("[Producer] Starting...")

    for i := 1; i <= 20; i++ {
        id, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamKey,
            Values: map[string]interface{}{
                "task_id":     fmt.Sprintf("TASK-%03d", i),
                "description": fmt.Sprintf("Process item %d", i),
                "priority":    (i % 3) + 1, // Priority 1-3
                "created_at":  time.Now().Unix(),
            },
        }).Result()
        if err != nil {
            log.Printf("[Producer] Error: %v", err)
            continue
        }
        fmt.Printf("[Producer] Added task %d (ID: %s)\n", i, id)
        time.Sleep(500 * time.Millisecond) // Task arrives every 500ms
    }
    fmt.Println("[Producer] Finished producing tasks")
}

func consumer(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string, wg *sync.WaitGroup, duration time.Duration) {
    defer wg.Done()
    fmt.Printf("[%s] Starting...\n", consumerName)

    timeout := time.After(duration)
    taskCount := 0

    for {
        select {
        case <-timeout:
            fmt.Printf("[%s] Timeout reached. Processed %d tasks.\n", consumerName, taskCount)
            return
        default:
            // Read messages from the group
            streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    groupName,
                Consumer: consumerName,
                Streams:  []string{streamKey, ">"},
                Count:    1,
                Block:    1 * time.Second,
            }).Result()

            if err != nil {
                if err == redis.Nil {
                    continue // No messages, keep waiting
                }
                log.Printf("[%s] Error: %v", consumerName, err)
                return
            }

            // Process messages
            for _, stream := range streams {
                for _, msg := range stream.Messages {
                    fmt.Printf("[%s] 🔄 Processing: %s - %s\n",
                        consumerName,
                        msg.Values["task_id"],
                        msg.Values["description"])

                    // Simulate task processing
                    processingTime := time.Duration(300+taskCount*50) * time.Millisecond
                    time.Sleep(processingTime)

                    // Acknowledge the message
                    err := rdb.XAck(ctx, streamKey, groupName, msg.ID).Err()
                    if err != nil {
                        log.Printf("[%s] Failed to ack %s: %v", consumerName, msg.ID, err)
                    } else {
                        fmt.Printf("[%s] ✅ Completed: %s (took %v)\n",
                            consumerName,
                            msg.Values["task_id"],
                            processingTime)
                        taskCount++
                    }
                }
            }
        }
    }
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "tasks:processing"
    groupName := "task-workers"

    // Clean up from previous runs
    rdb.Del(ctx, streamKey)

    // Create consumer group
    err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0").Err()
    if err != nil {
        log.Fatalf("Failed to create group: %v", err)
    }
    fmt.Println("Created consumer group:", groupName)

    var wg sync.WaitGroup

    // Start producer
    wg.Add(1)
    go producer(ctx, rdb, streamKey, &wg)

    // Start 3 consumers
    wg.Add(3)
    go consumer(ctx, rdb, streamKey, groupName, "Worker-1", &wg, 15*time.Second)
    go consumer(ctx, rdb, streamKey, groupName, "Worker-2", &wg, 15*time.Second)
    go consumer(ctx, rdb, streamKey, groupName, "Worker-3", &wg, 15*time.Second)

    // Wait for all goroutines to finish
    wg.Wait()

    // Check final stream state
    length, _ := rdb.XLen(ctx, streamKey).Result()
    pending, _ := rdb.XPending(ctx, streamKey, groupName).Result()

    fmt.Println("\n=== Final Statistics ===")
    fmt.Printf("Stream length: %d messages\n", length)
    fmt.Printf("Pending messages: %d\n", pending.Count)
    fmt.Println("\nAll tasks processed successfully!")
}

Exercise 3: Stream Trimming and Monitoring (15 minutes)

Objective: Implement automatic stream trimming and monitoring

Challenge: Create a monitoring system that:

  1. Produces events continuously for 30 seconds
  2. Implements automatic trimming to keep only the last 100 entries
  3. Monitors and reports stream statistics every 5 seconds
  4. Handles pending messages from failed consumers

Extension Ideas:

  • Add memory usage estimation based on entry count
  • Implement different trimming strategies (MAXLEN vs MINID)
  • Create alerts when pending messages exceed a threshold
  • Track processing latency and throughput

Starter Code:

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
)

func monitorStream(ctx context.Context, rdb *redis.Client, streamKey, groupName string) {
    // TODO: Implement monitoring that reports:
    // - Stream length
    // - Number of consumers
    // - Pending message count
    // - Consumer lag
}

func producerWithTrimming(ctx context.Context, rdb *redis.Client, streamKey string) {
    // TODO: Add events with MAXLEN trimming to keep stream size bounded
}

func main() {
    // TODO: Implement complete solution
}

Solution:

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

func monitorStream(ctx context.Context, rdb *redis.Client, streamKey, groupName string, done chan bool) {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    fmt.Println("[Monitor] Starting stream monitoring...")

    for {
        select {
        case <-done:
            fmt.Println("[Monitor] Stopping...")
            return
        case <-ticker.C:
            // Get stream info
            length, err := rdb.XLen(ctx, streamKey).Result()
            if err != nil {
                log.Printf("[Monitor] Error getting length: %v", err)
                continue
            }

            // Get pending info
            pending, err := rdb.XPending(ctx, streamKey, groupName).Result()
            if err != nil && err != redis.Nil {
                log.Printf("[Monitor] Error getting pending: %v", err)
                continue
            }

            // Get group info using XINFO GROUPS
            groups, err := rdb.XInfoGroups(ctx, streamKey).Result()
            if err != nil {
                log.Printf("[Monitor] Error getting groups: %v", err)
                continue
            }

            fmt.Println("\n=== Stream Statistics ===")
            fmt.Printf("Timestamp: %s\n", time.Now().Format("15:04:05"))
            fmt.Printf("Stream length: %d entries\n", length)
            
            if pending != nil {
                fmt.Printf("Pending messages: %d\n", pending.Count)
                if pending.Count > 0 {
                    fmt.Printf("Pending range: %s to %s\n", pending.Lower, pending.Upper)
                    fmt.Println("Consumers with pending:")
                    for consumer, count := range pending.Consumers {
                        fmt.Printf("  - %s: %d messages\n", consumer, count)
                    }
                }
            }

            if len(groups) > 0 {
                fmt.Printf("\nConsumer Groups: %d\n", len(groups))
                for _, group := range groups {
                    fmt.Printf("  Group: %s\n", group.Name)
                    fmt.Printf("    Consumers: %d\n", group.Consumers)
                    fmt.Printf("    Pending: %d\n", group.Pending)
                    fmt.Printf("    Last delivered: %s\n", group.LastDeliveredID)
                }
            }

            // Estimate memory usage (rough estimate: 50 bytes per entry + data)
            estimatedMemory := length * 100 // Very rough estimate
            fmt.Printf("\nEstimated memory: ~%d KB\n", estimatedMemory/1024)
            fmt.Println("========================")
        }
    }
}

func producerWithTrimming(ctx context.Context, rdb *redis.Client, streamKey string, duration time.Duration) {
    fmt.Println("[Producer] Starting with auto-trimming...")
    
    timeout := time.After(duration)
    count := 0

    for {
        select {
        case <-timeout:
            fmt.Printf("[Producer] Finished. Total produced: %d events\n", count)
            return
        default:
            // XADD with MAXLEN to trim automatically
            id, err := rdb.XAdd(ctx, &redis.XAddArgs{
                Stream: streamKey,
                MaxLen: 100, // Keep only last 100 entries
                Approx: true, // Use approximate trimming for better performance
                Values: map[string]interface{}{
                    "event_num": count,
                    "timestamp": time.Now().Unix(),
                    "data":      fmt.Sprintf("Event data %d", count),
                },
            }).Result()
            if err != nil {
                log.Printf("[Producer] Error: %v", err)
                continue
            }
            
            count++
            if count%10 == 0 {
                fmt.Printf("[Producer] Produced %d events (latest: %s)\n", count, id)
            }
            
            time.Sleep(200 * time.Millisecond) // 5 events per second
        }
    }
}

func consumerWithRecovery(ctx context.Context, rdb *redis.Client, streamKey, groupName, consumerName string, duration time.Duration) {
    fmt.Printf("[%s] Starting...\n", consumerName)
    
    timeout := time.After(duration)
    processed := 0

    for {
        select {
        case <-timeout:
            fmt.Printf("[%s] Stopping. Processed %d messages\n", consumerName, processed)
            return
        default:
            // First check for pending messages
            pendingStreams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    groupName,
                Consumer: consumerName,
                Streams:  []string{streamKey, "0"},
                Count:    5,
                Block:    0,
            }).Result()

            if err == nil && len(pendingStreams) > 0 {
                for _, stream := range pendingStreams {
                    for _, msg := range stream.Messages {
                        fmt.Printf("[%s] 🔄 Recovering pending: %v\n", consumerName, msg.Values["event_num"])
                        time.Sleep(100 * time.Millisecond)
                        rdb.XAck(ctx, streamKey, groupName, msg.ID)
                        processed++
                    }
                }
            }

            // Read new messages
            streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    groupName,
                Consumer: consumerName,
                Streams:  []string{streamKey, ">"},
                Count:    5,
                Block:    1 * time.Second,
            }).Result()

            if err != nil {
                if err != redis.Nil {
                    log.Printf("[%s] Error: %v", consumerName, err)
                }
                continue
            }

            for _, stream := range streams {
                for _, msg := range stream.Messages {
                    // Process message
                    time.Sleep(150 * time.Millisecond)
                    rdb.XAck(ctx, streamKey, groupName, msg.ID)
                    processed++
                    
                    if processed%10 == 0 {
                        fmt.Printf("[%s] Processed %d messages\n", consumerName, processed)
                    }
                }
            }
        }
    }
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    defer rdb.Close()

    streamKey := "events:monitored"
    groupName := "monitoring-group"

    // Clean up
    rdb.Del(ctx, streamKey)

    // Create consumer group
    err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0").Err()
    if err != nil {
        log.Fatalf("Failed to create group: %v", err)
    }

    done := make(chan bool)

    // Start monitoring
    go monitorStream(ctx, rdb, streamKey, groupName, done)

    // Start producer with trimming
    go producerWithTrimming(ctx, rdb, streamKey, 30*time.Second)

    // Start consumers
    go consumerWithRecovery(ctx, rdb, streamKey, groupName, "Consumer-A", 30*time.Second)
    go consumerWithRecovery(ctx, rdb, streamKey, groupName, "Consumer-B", 30*time.Second)

    // Run for 30 seconds
    time.Sleep(32 * time.Second)
    close(done)

    time.Sleep(1 * time.Second)
    fmt.Println("\n=== Exercise Complete ===")
}

7. Common Pitfalls and Troubleshooting (10 minutes)

Frequent Mistakes

  1. Problem: Messages are not distributed among consumers in a group Cause: All consumers are reading with ID “0” instead of “>” Solution: Use “>” to get only new, undelivered messages. Use “0” only for recovering pending messages.
// Wrong - all consumers see all messages
rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Streams: []string{streamKey, "0"}, // ❌ Returns already delivered messages
})

// Correct - each consumer gets unique messages
rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Streams: []string{streamKey, ">"}, // ✅ Returns only undelivered messages
})
  1. Problem: Consumer group doesn’t exist error (NOGROUP) Cause: Trying to read before creating the consumer group Solution: Always create the consumer group before any consumer reads from it, and handle the “already exists” error gracefully.
// Always create group first
err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0").Err()
if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
    log.Fatal(err)
}
  1. Problem: Stream grows indefinitely, consuming too much memory Cause: Not implementing trimming strategy Solution: Use MAXLEN or MINID when adding entries, or periodically trim with XTRIM.
// With automatic trimming
rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: streamKey,
    MaxLen: 10000,      // Keep only last 10k entries
    Approx: true,       // Use ~ for better performance
    Values: data,
})

// Or manual trimming
rdb.XTrimMaxLen(ctx, streamKey, 10000)
  1. Problem: Messages are processed multiple times Cause: Acknowledging messages before actually processing them, then crashing during processing Solution: Always acknowledge after successful processing, not before.
// Wrong order - message lost if processing fails
rdb.XAck(ctx, streamKey, groupName, msg.ID)  // ❌ Too early
processMessage(msg)  // If this fails, message is already acked

// Correct order - message stays pending if processing fails
if processMessage(msg) {
    rdb.XAck(ctx, streamKey, groupName, msg.ID)  // ✅ Only ack after success
}
  1. Problem: Consumer crashes and messages are stuck in pending state Cause: Not implementing claim mechanism for stale messages Solution: Periodically check for and claim messages that have been pending too long.
// Check for stale messages
pending, _ := rdb.XPendingExt(ctx, &redis.XPendingExtArgs{
    Stream: streamKey,
    Group:  groupName,
    Start:  "-",
    End:    "+",
    Count:  100,
})

for _, msg := range pending {
    if msg.Idle > 30*time.Second {
        // Claim stale message
        rdb.XClaim(ctx, &redis.XClaimArgs{
            Stream:   streamKey,
            Group:    groupName,
            Consumer: "recovery-consumer",
            MinIdle:  30 * time.Second,
            Messages: []string{msg.ID},
        })
    }
}

Debugging Tips

Use Redis CLI for inspection:

# Check stream contents
redis-cli XRANGE mystream - + COUNT 10

# Check consumer group info
redis-cli XINFO GROUPS mystream

# Check pending messages
redis-cli XPENDING mystream mygroup

# Get detailed pending info
redis-cli XPENDING mystream mygroup - + 10

# Manually acknowledge a message
redis-cli XACK mystream mygroup 1609459200000-0

Enable logging in your application:

// Log all stream operations
fmt.Printf("[DEBUG] XReadGroup: stream=%s, group=%s, consumer=%s\n", 
    streamKey, groupName, consumerName)

Monitor Redis with INFO:

# Check Redis memory usage
redis-cli INFO memory

# Check stream-related stats
redis-cli INFO stream

8. Best Practices and Professional Tips (5 minutes)

Industry Standards

  1. Use Consumer Groups for Reliability

    • Even with a single consumer, use consumer groups for progress tracking
    • Consumer groups provide at-least-once delivery guarantees
    • Enable easy horizontal scaling by adding consumers

  2. Implement Proper Error Handling

    • Always check for pending messages on consumer startup
    • Implement message claiming for failure recovery
    • Use exponential backoff for failed message processing
    • Log failed messages to a dead-letter queue after max retries

  3. Memory Management

    • Set appropriate MAXLEN limits based on your retention needs
    • Use approximate trimming (~) for better performance
    • Monitor memory usage with Redis INFO commands
    • Consider time-based trimming with MINID for time-series data

  4. Message Processing Patterns

    • Make message processing idempotent (safe to process multiple times)
    • Acknowledge only after successful processing
    • Use transactions or pipelines to batch operations
    • Implement circuit breakers for downstream service failures

  5. Monitoring and Observability

    • Track stream length, pending messages, and consumer lag
    • Alert on growing pending message counts
    • Monitor processing latency and throughput
    • Use Redis’s built-in XINFO commands for diagnostics

  6. Performance Optimization

    • Use blocking reads instead of polling
    • Batch message processing when possible
    • Use pipelines for multiple Redis commands
    • Consider connection pooling for high throughput

Code Organization Tips

// Encapsulate stream operations in a struct
type StreamProcessor struct {
    client      *redis.Client
    streamKey   string
    groupName   string
    consumerName string
}

func (sp *StreamProcessor) Process(ctx context.Context, handler func(msg redis.XMessage) error) error {
    // Centralized processing logic with error handling
}

// Use dependency injection for testability
type MessageHandler interface {
    Handle(msg redis.XMessage) error
}

// Implement retry logic as middleware
func withRetry(handler MessageHandler, maxRetries int) MessageHandler {
    return &retryHandler{
        handler: handler,
        maxRetries: maxRetries,
    }
}

Next Steps in Your Learning Journey

  1. Explore Advanced Patterns

    • Implement exactly-once semantics with deduplication
    • Build complex event processing with multiple streams
    • Create stream aggregation and windowing functions

  2. Integration with Other Systems

    • Connect streams with Apache Kafka using connectors
    • Build ETL pipelines with Redis Streams
    • Implement CQRS patterns with event sourcing

  3. Production Deployment

    • Set up Redis Sentinel or Cluster for high availability
    • Implement monitoring with Prometheus and Grafana
    • Use Redis Enterprise for multi-datacenter replication

  4. Performance Tuning

    • Benchmark your stream workloads
    • Optimize serialization (consider msgpack or protobuf)
    • Profile memory usage and identify bottlenecks

9. Summary and Review (5 minutes)

Key Takeaways

  1. Redis Streams provide persistent, reliable message streaming with consumer groups for distributed processing
  2. Consumer groups enable at-least-once delivery through pending entries and acknowledgments
  3. Stream IDs are time-based (millisecond-timestamp-sequence), enabling efficient time-range queries
  4. Always acknowledge messages after processing, not before, to prevent data loss
  5. Implement trimming strategies to prevent unbounded memory growth
  6. Use XClaim to recover from consumer failures by reassigning stale pending messages

Quick Review Questions

  1. What’s the difference between reading with ID “>” vs “0” in a consumer group?

    • “>” returns only new, undelivered messages (for normal processing)
    • “0” returns all pending messages for this consumer (for recovery)

  2. When should you use Redis Streams vs Pub/Sub?

    • Use Streams when you need: persistence, message history, consumer groups, at-least-once delivery
    • Use Pub/Sub when you need: fire-and-forget, real-time broadcast, minimal latency, no persistence

  3. How do you handle a consumer that crashes with pending messages?

    • Another consumer uses XPENDING to find stale messages
    • Uses XCLAIM to take ownership of those messages
    • Processes and acknowledges them

  4. What happens if you don’t trim a stream?

    • Stream grows indefinitely, consuming memory
    • Can lead to out-of-memory errors
    • Performance degradation as stream size increases

For Instructors

Teaching Tips

  • Start with simple producer-consumer before introducing consumer groups
  • Use visual diagrams to explain consumer group mechanics
  • Have students experiment with crashing consumers to see pending message recovery
  • Compare Streams with other message systems (Kafka, RabbitMQ) students might know
  • Emphasize the importance of acknowledgment timing with live demos
  • Show real Redis INFO output to make monitoring concrete

Common Student Questions

  • “Why not just use Kafka?” - Redis Streams is simpler to operate, lower latency for smaller workloads, and integrates seamlessly if you’re already using Redis
  • “What’s the max throughput?” - Typically 100k-500k messages/sec depending on message size and hardware
  • “Can I have multiple groups on one stream?” - Yes! Different applications can independently consume the same stream
  • “What happens if Redis restarts?” - Streams are persisted (with appropriate Redis persistence config), consumer groups and pending entries are preserved

Assessment Ideas

Quiz Questions:

  1. Write code to create a consumer group starting from the end of a stream
  2. Explain the lifecycle of a message from XADD to XACK
  3. Design a solution for processing exactly-once (hint: deduplication)

Project Rubric:

  • Correct consumer group implementation (30%)
  • Proper error handling and message acknowledgment (25%)
  • Stream trimming strategy (15%)
  • Pending message recovery (20%)
  • Code quality and documentation (10%)

Peer Review Activity:

  • Students review each other’s consumer implementations
  • Check for proper acknowledgment timing
  • Verify error handling for edge cases

Adaptation Notes

For Beginners:

  • Spend more time on basic XADD/XREAD before consumer groups
  • Use simpler examples without goroutines initially
  • Provide more scaffolding in exercises

For Advanced Students:

  • Challenge them to implement exactly-once semantics
  • Discuss trade-offs with other message brokers
  • Explore integration with stream processing frameworks
  • Benchmark and optimize for maximum throughput

Online vs In-Person:

  • Online: Use breakout rooms for exercise collaboration
  • In-person: Live coding on projector with student participation
  • Both: Encourage experiments in Redis CLI alongside Go code

Time Management:

  • Minimum viable lesson: 60 minutes (skip Exercise 3, simplified examples)
  • Standard lesson: 90 minutes (as written)
  • Extended workshop: 120 minutes (add comparative analysis with Kafka/RabbitMQ)