Redis Streams and Pub Sub Used From Go Programs Compared

Redis offers two ways to move messages between processes, and they are not alike. Pub/Sub broadcasts to whoever is listening right now and forgets. Streams are an append-only log with consumer groups, acknowledgements and the ability to reclaim work from a consumer that died. Choosing between them is mostly a question of what you can afford to lose. This article runs both from Go 1.25.6 with go-redis v9.17.0 and Redis 7, shows what each does to a message that arrives at an awkward time, and goes through the stream commands you need in practice.

It belongs to the database cornerstone and builds on the go-redis article. For a queue stored in PostgreSQL, see a Postgres job queue with SKIP LOCKED.

The lab

package main

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

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

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:56379"})
	defer rdb.Close()
	_ = rdb.FlushDB(ctx).Err()

	// 1. Pub/Sub keeps nothing: a message sent before anyone subscribes is gone.
	rdb.Publish(ctx, "events", "sent before the subscriber existed")
	sub := rdb.Subscribe(ctx, "events")
	if _, err := sub.Receive(ctx); err != nil {
		log.Fatal(err)
	}
	ch := sub.Channel()
	rdb.Publish(ctx, "events", "sent after subscribing")
	select {
	case m := <-ch:
		fmt.Printf("1. pub/sub delivered only: %q\n", m.Payload)
	case <-time.After(time.Second):
		fmt.Println("1. nothing delivered")
	}
	_ = sub.Close()

	// 2. A stream keeps entries. A group tracks what each consumer has read and not acknowledged.
	for i := 1; i <= 6; i++ {
		rdb.XAdd(ctx, &redis.XAddArgs{Stream: "orders", Values: map[string]any{"order": i}})
	}
	_ = rdb.XGroupCreateMkStream(ctx, "orders", "billing", "0").Err()
	n, _ := rdb.XLen(ctx, "orders").Result()
	fmt.Println("2. entries in the stream:", n, "(added before the group existed, read from position 0)")

	read := func(consumer string, count int64) []redis.XMessage {
		res, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{Group: "billing", Consumer: consumer, Streams: []string{"orders", ">"}, Count: count, Block: 100 * time.Millisecond}).Result()
		if errors.Is(err, redis.Nil) || len(res) == 0 {
			return nil
		}
		return res[0].Messages
	}
	ids := func(ms []redis.XMessage) (out []string) {
		for _, m := range ms {
			out = append(out, fmt.Sprint(m.Values["order"]))
		}
		return
	}

	// 3. Consumer c1 reads three entries, acknowledges one and then "crashes".
	got := read("c1", 3)
	_ = rdb.XAck(ctx, "orders", "billing", got[0].ID).Err()
	fmt.Println("3. c1 read orders", ids(got), "and acknowledged order", got[0].Values["order"])
	pend, _ := rdb.XPending(ctx, "orders", "billing").Result()
	fmt.Printf("   pending for the group: %d, owned by %v\n", pend.Count, pend.Consumers)

	// 4. c2 keeps reading new entries, and a different command claims what c1 abandoned.
	fresh := read("c2", 10)
	fmt.Println("4. c2 read the rest:", ids(fresh))
	for _, m := range fresh {
		_ = rdb.XAck(ctx, "orders", "billing", m.ID).Err()
	}
	time.Sleep(150 * time.Millisecond)
	claimed, _, err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{Stream: "orders", Group: "billing", Consumer: "c2", MinIdle: 100 * time.Millisecond, Start: "0", Count: 10}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("   after 150ms, XAUTOCLAIM moved c1's unacknowledged orders to c2:", ids(claimed))
	for _, m := range claimed {
		_ = rdb.XAck(ctx, "orders", "billing", m.ID).Err()
	}
	pend, _ = rdb.XPending(ctx, "orders", "billing").Result()
	fmt.Println("   pending after c2 acknowledged everything:", pend.Count)

	// 5. The stream still holds the history. Trim it so it does not grow without bound.
	n, _ = rdb.XLen(ctx, "orders").Result()
	trimmed, _ := rdb.XTrimMaxLen(ctx, "orders", 2).Result()
	left, _ := rdb.XLen(ctx, "orders").Result()
	fmt.Printf("5. acknowledged entries are still stored (%d); XTRIM MAXLEN 2 removed %d, %d left\n", n, trimmed, left)

	// 6. A second group reads the same stream independently, from the start.
	_ = rdb.XGroupCreate(ctx, "orders", "analytics", "0").Err()
	res, _ := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{Group: "analytics", Consumer: "a1", Streams: []string{"orders", ">"}, Count: 10, Block: 100 * time.Millisecond}).Result()
	fmt.Println("6. a new group sees the entries that remain:", ids(res[0].Messages))
}

1. pub/sub delivered only: "sent after subscribing"
2. entries in the stream: 6 (added before the group existed, read from position 0)
3. c1 read orders [1 2 3] and acknowledged order 1
   pending for the group: 2, owned by map[c1:2]
4. c2 read the rest: [4 5 6]
   after 150ms, XAUTOCLAIM moved c1's unacknowledged orders to c2: [2 3]
   pending after c2 acknowledged everything: 0
5. acknowledged entries are still stored (6); XTRIM MAXLEN 2 removed 4, 2 left
6. a new group sees the entries that remain: [5 6]

Pub/Sub keeps nothing

The program publishes a message to the channel events before any subscriber exists, then subscribes, publishes another and reads from the subscription's channel. Only sent after subscribing arrived. The first message went to zero listeners, and Redis does not keep it for later. This is the whole model: PUBLISH returns the number of subscribers that received the message, and a subscriber that is slow, disconnected or restarting misses everything sent meanwhile. There are no acknowledgements and no replay.

That is fine for what it fits: cache invalidation hints, "something changed, go look" notifications, live presence, fan out to connected websockets, anywhere that a missed message is harmless or recovered by a later one. It is the wrong choice for work that must happen, such as an order to bill. Note also that a subscribed connection is dedicated to that purpose and a client with a slow reader can be disconnected by the server when its output buffer grows too large.

Streams store entries

A stream is a log. XADD orders * order 1 appends an entry with an id made of a timestamp and a sequence, and entries stay until you remove them. The lab added six, created a consumer group named billing on the stream with XGroupCreateMkStream(..., "0"), which starts the group at the beginning, and confirmed XLEN was 6. The entries were written before the group existed and the group could still read them, which Pub/Sub cannot do.

A consumer group is a cursor shared by cooperating consumers. XREADGROUP GROUP billing c1 STREAMS orders > asks for entries nobody in the group has been given yet, and Redis hands each one to exactly one consumer, so several consumers split the work. Handing an entry over is not completing it: the entry goes onto the group's pending entries list against that consumer's name until it is acknowledged with XACK.

Acknowledge, crash, reclaim

The lab simulates a failure. Consumer c1 read three entries (orders 1, 2 and 3), acknowledged the first and then stopped without handling the others, as a process killed mid job would. XPENDING on the group reported 2 pending entries, owned by map[c1:2]. Those two are neither lost nor delivered again on their own: Redis is waiting for c1 to acknowledge them.

Meanwhile c2 read the rest (orders 4, 5 and 6) and acknowledged them. Then the program waited 150 milliseconds and called XAutoClaim with MinIdle: 100ms. It returned orders 2 and 3, now owned by c2. XAUTOCLAIM finds pending entries that have been idle for at least the given time, whoever owns them, and reassigns them to the caller. After c2 acknowledged those, XPENDING showed 0.

This is the recovery loop for a consumer group, and every real consumer needs it:

1. Read new entries with > and process them.
2. Acknowledge each entry after it has been handled and its effects are durable.
3. Periodically run XAUTOCLAIM with an idle time longer than the slowest normal handling, and process what it returns as you would new entries.

The idle time is a judgment. Too short and a slow consumer's entry is taken while it is still working, so two consumers process it; too long and a crashed consumer's work waits. Because a reclaim can duplicate work, handlers must be idempotent: delivery is at least once, the same property as the outbox in the outbox pattern. Each entry also has a delivery count, available from XPENDING in its extended form, so you can detect a poison entry that keeps failing and move it to a dead letter stream after a few attempts.

Entries are not removed on acknowledge

After everything was acknowledged, XLEN was still 6. Acknowledging removes the entry from the pending list, not from the stream. A stream grows until you trim it. XTRIM orders MAXLEN 2 removed 4 entries and left 2; MAXLEN ~ 1000 with the approximate flag trims in efficient chunks and is the usual setting, and XADD accepts the same option so the stream trims as it grows. The alternative is MINID, which trims by age.

Keeping the history is a feature when a new consumer needs it. The last step created a second group, analytics, on the same stream. It read the entries that remained, orders 5 and 6, independently of the billing group's progress: each group has its own cursor, so one stream can feed several independent consumers, like topics in Kafka with partitions replaced by consumers sharing a group. If you trim too soon, a new group cannot start from the beginning, so decide the retention with that in mind.

Choosing

Redis durability is the last caveat. A stream is in memory, written to disk by the append only file or snapshots depending on configuration, and with the default settings a crash can lose the last second of entries. For a queue of work that must not be lost, make sure appendonly yes and an appendfsync policy match your tolerance, or keep the source of truth in a database and use the stream as the transport.

Go specifics

• Use a separate client or connection for blocking reads. XREADGROUP with Block ties up a pooled connection for the block time. Size the pool for the number of consumer goroutines, as in go-redis.
• Respect deadlines. A blocking read needs ContextTimeoutEnabled for a context to cancel it, and a ReadTimeout longer than the block time, or the client times out first.
• Handle redis.Nil. A read that times out with no entries returns redis.Nil, which is not an error; the lab's read helper treats it that way.
• Create groups idempotently. XGroupCreateMkStream returns a BUSYGROUP error when the group exists, which a start-up routine should ignore.
• Acknowledge in the same code path as the effect, and do not acknowledge before the work is durable.

The commands are documented in the Redis pages on Streams and Pub/Sub, which include the delivery guarantees stated above.