Real-time with Redis Streams in Go

#go #webdev #programming #beginners

Hello, I'm Maneshwar. I'm working on FreeDevTools online currently building **one place for all dev tools, cheat codes, and TLDRs* — a free, open-source hub where developers can quickly find and use tools without any hassle of searching all over the internet.

Real-time with Redis Streams in Go

Redis Stre
ams give you Kafka-like message queues with Redis simplicity. Whether you’re building real-time analytics, background job pipelines, or chat systems, Redis Streams can help.

In this post, we’ll cover:

What is a Redis Stream?
Setting up Redis
Writing to a Stream in Go
Reading from a Stream in Go
Using Consumer Groups
Stream Configuration Parameters

What is a Redis Stream?

A Redis Stream is an append-only log data structure where each entry has a unique ID and a set of key-value fields.

You write using XADD, read using XREAD, and scale consumption using consumer groups.

# Example
XADD mystream * name Alice action login

Setup

Install Redis:

sudo apt install redis
redis-server

Install Go Redis client:

go get github.com/redis/go-redis/v9

Writing to a Stream in Go

package main

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

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

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    streamKey := "events"
    args := &redis.XAddArgs{
        Stream: streamKey,
        MaxLen: 1000,
        Approx: true,
        Values: map[string]interface{}{
            "user":   "maneshwar",
            "action": "upload",
            "time":   time.Now().Format(time.RFC3339),
        },
    }
    id, err := rdb.XAdd(ctx, args).Result()
    if err != nil {
        log.Fatalf("XAdd failed: %v", err)
    }

    fmt.Printf("Written to stream with ID: %s\n", id)
}

Reading from a Stream in Go

for {
    streams, err := rdb.XRead(ctx, &redis.XReadArgs{
        Streams: []string{"events", "$"},
        Block:   0,
        Count:   1,
    }).Result()
    if err != nil {
        log.Fatalf("XRead failed: %v", err)
    }

    for _, stream := range streams {
        for _, msg := range stream.Messages {
            fmt.Printf("ID: %s, Values: %v\n", msg.ID, msg.Values)
        }
    }
}

Using Consumer Groups

err := rdb.XGroupCreateMkStream(ctx, "events", "workers", "$").Err()
if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
    log.Fatal(err)
}

res, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "workers",
    Consumer: "worker-1",
    Streams:  []string{"events", ">"},
    Block:    time.Second * 5,
    Count:    10,
}).Result()

for _, stream := range res {
    for _, msg := range stream.Messages {
        fmt.Printf("Group message: %s => %v\n", msg.ID, msg.Values)
        rdb.XAck(ctx, "events", "workers", msg.ID)
    }
}

Stream Configuration Parameters

Trimming the Stream

Exact Trimming:

XADD mystream MAXLEN 1000 * field1 val1

Approximate Trimming (better performance):

XADD mystream MAXLEN ~ 1000 * field1 val1

Or in Go:

args := &redis.XAddArgs{
    Stream:   "events",
    MaxLen:   1000,
    Approx:   true,
    Values:   map[string]interface{}{
        "user": "bob",
        "action": "logout",
    },
}
rdb.XAdd(ctx, args)

Scheduled Trim Example:

length, _ := rdb.XLen(ctx, "events").Result()
if length > 2000 {
    rdb.XTrim(ctx, "events", &redis.XTrimArgs{MaxLenApprox: 1000})
}

Memory Efficiency

Tune these Redis configs for stream node sizes:

CONFIG SET stream-node-max-bytes 4096
CONFIG SET stream-node-max-entries 100

Helps approximate trimming work better and keeps memory predictable.

Persistence with PERSIST flag

Use PERSIST in Redis CLI to force entry persistence:

XADD mystream PERSIST MAXLEN ~ 500 * field val

(Current Go clients may not support this yet.)

Summary

Feature	Redis CLi	Go (`go-redis`)
Exact trim	`XADD MAXLEN 1000`	`XAddArgs{MaxLen: 1000}`
Approximate trim (`~`)	`XADD MAXLEN ~ 1000`	`XAddArgs{MaxLen:1000,Approx:true}`
Periodic trim	`XTRIM MAXLEN ~ 1000`	`XTrimArgs{MaxLenApprox:1000}`
Trim by ID	`XTRIM MINID 1605...-0`	`XTrimArgs{MinID:"1605...-0"}`
Memory tuning	`CONFIG SET ...`	via CLI
Persistence control	`XADD PERSIST ...`	not yet exposed in Go clients