Introduction to Redis Streams with Go

Welcome back! In this lesson, we will dive into Redis Streams — a powerful feature used for processing streams of data. This lesson will guide you through the basics and show you how Redis Streams can be essential for high-performance applications.

In this lesson, we'll explore streams in Redis and how they can be used to handle continuous data flows. You'll learn how to create streams, add events to them, and read events from them.

Streams in Redis are data structures that follow the FIFO (First In, First Out) principle. Each entry in a stream is stored with a unique ID, which by default includes the current timestamp, but can be customized when adding events. Streams can efficiently handle continuous flows of data, making them ideal for use cases like chat applications, monitoring systems, or user activity tracking.

Redis Streams are particularly valuable in concurrent scenarios, where multiple consumers need to efficiently process real-time data. While we don't cover concurrent consumption in detail in this lesson, it's important to understand this crucial aspect of streams. Consider these real-world scenarios where Redis Streams can be beneficial:

• Chat Applications: Ideal for real-time message handling.
• Monitoring Systems: Useful for processing logs and events.
• User Activity Tracking: Tracks user actions in real-time.

Let's dive into the details!

To add an event to a stream, use the XAdd command.
To read events from a stream, use the XRange command.

Let's see how these commands work in practice using Go.

package main

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

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

    defer client.Close()

    // Adding events to the stream
    client.XAdd(ctx, &redis.XAddArgs{
        Stream: "mystream",  // Name of the stream where the event will be added
        Values: map[string]interface{}{"event": "login", "user": "Alice"},  // Key-value pairs representing the event data
        ID:     "*",  // Unique identifier for the entry, "*" auto-generates based on the timestamp
    })
    client.XAdd(ctx, &redis.XAddArgs{
        Stream: "mystream",
        Values: map[string]interface{}{"event": "purchase", "user": "Bob", "amount": "100"},
        ID:     "*",
    })
    client.XAdd(ctx, &redis.XAddArgs{
        Stream: "mystream",
        Values: map[string]interface{}{"event": "add_to_cart", "user": "Alice", "product": "laptop"},
        ID:     "*",
    })

    // Reading events from the stream
    messages, err := client.XRange(ctx, "mystream", "-", "+").Result()
    if err != nil {
        fmt.Println("Error reading stream:", err)
        return
    }

    fmt.Printf("Stream messages: %v\n", messages)

    if len(messages) > 0 {
        firstMessage := messages[0].Values
        fmt.Printf("First message: %v\n", firstMessage) // {"event": "login", "user": "Alice"}
    }
}

The above code snippet demonstrates how to add events to a Redis stream called mystream using XAdd. Each event contains key-value pairs representing different actions by users.

The code reads messages from mystream and prints them. The "-" and "+" arguments to XRange indicate reading messages from the beginning to the end of the stream.

Notice that to access a single message from the events slice, you can use the messages[i].Values property, which contains the actual event data. In the example above, we access the first message and print it to the console.