Skip to content

Reading Redis Streams with Consumer Groups

Most services that need a work queue already run Redis, and Redis Streams turn it into one with per-message acknowledgement, consumer groups, replay and recovery from dead workers — no extra service to operate. The mechanism that makes it reliable is the pending entry list: every message delivered to a group member stays recorded against that member until it is acknowledged, so a worker that crashes leaves its work visible and claimable rather than lost. That is also where the bugs live, because a consumer that forgets to acknowledge, or never claims abandoned entries, accumulates a backlog nothing will ever process. This guide builds a group consumer against a live Redis 8 with redis.asyncio, measuring each step.

Prerequisites

A message through a consumer group 5 stages from XADD to XACK. A message through a consumer group XADD append, with a time id XREADGROUP delivered to one member pending entry list until acknowledged handler runs your processing XACK or it stays claimable The pending entry list is the whole reliability mechanism: nothing leaves it unacknowledged.

1. Create the group and write with trimming

The group must exist before anyone reads; mkstream=True creates the stream at the same time:

r = redis.from_url(REDIS_URL, decode_responses=True)
try:
    await r.xgroup_create("events", "workers", id="0", mkstream=True)
except redis.ResponseError as exc:
    if "BUSYGROUP" not in str(exc):                    # already exists: fine
        raise

id="0" starts the group at the beginning of the stream; id="$" starts it at the end, delivering only messages added after creation. Choose deliberately — a group created with $ silently skips everything already queued.

Writes should always trim, or the stream grows until Redis runs out of memory:

async with r.pipeline(transaction=False) as pipe:
    for event in events:
        pipe.xadd("events", event, maxlen=10_000, approximate=True)   # the ~ form
    ids = await pipe.execute()

1,000 pipelined XADDs took 9 ms. approximate=True emits MAXLEN ~, which lets Redis trim on macro-node boundaries — much cheaper than the exact form, and the reason to prefer it unless the bound must be precise. The returned ids are millisecond timestamps with a sequence suffix, such as 1789674080894-0, and they sort lexicographically in time order.

Verify: XLEN stabilises near your MAXLEN under sustained writes.

2. Read as a group member

XREADGROUP delivers each message to exactly one member of the group and records it as pending:

response = await r.xreadgroup("workers", consumer_name, {"events": ">"},
                              count=100, block=5000)
for stream, messages in response or []:
    for message_id, fields in messages:
        await handle(fields)
        await r.xack("events", "workers", message_id)

Reading 100 messages took 1.3 ms. The > is significant: it means "messages never delivered to this group". Passing 0 instead replays this consumer's own pending entries, which is how a worker recovers its own unfinished work after a restart — a different operation that looks almost identical.

block=5000 waits up to five seconds for new messages rather than polling, and returns None on timeout. Verified: a second member reading with > received an entirely disjoint set from the first, which is the group property you are relying on.

The consumer name should be stable per worker — the pod name, or the hostname plus a slot number — because pending entries are tracked against it and a random name per restart strands them under a consumer that no longer exists.

Verify: two members of the same group never receive the same message id.

The commands a consumer needs A grid of 5 rows by 2 columns. The commands a consumer needs command does watch out for XADD ... MAXLEN ~ n append and trim without it the stream grows forever XREADGROUP ... > new messages for this member 0 replays your own pending XACK clears a pending entry forgetting it leaks pending entries XAUTOCLAIM min-idle-time takes over abandoned work raises times_delivered XPENDING who holds what, for how long the queue depth you alert on Measured: 1,000 pipelined XADDs took 9 ms; a 100-message group read took 1.3 ms.

3. Acknowledge, and watch what happens when you do not

XACK removes the entry from the pending list. Acking 50 of 100 messages left exactly 50 pending, confirmed by XPENDING:

summary = await r.xpending("events", "workers")
summary["pending"]                                     # 50

Unacknowledged entries are not lost — that is the point — but they are also not retried automatically. They sit in the pending list until someone claims them, so a consumer that processes successfully and forgets to XACK builds an ever-growing pending list while appearing to work perfectly. Alert on XPENDING depth, not only on stream length.

XPENDING with a range gives the detail needed for diagnosis:

await r.xpending_range("events", "workers", min="-", max="+", count=10)
# [{'message_id': '...-0', 'consumer': 'worker-3', 'time_since_delivered': ..., 'times_delivered': 2}, ...]

times_delivered is your poison-message detector: an entry whose delivery count keeps climbing is failing every time, and belongs in a dead-letter stream rather than in the rotation. Redis has no dead-lettering of its own, so that is an XADD to another stream followed by an XACK on the original.

Verify: the pending count returns to zero when the workers are idle.

4. Claim the work of a crashed worker

XAUTOCLAIM transfers entries that have been pending longer than a threshold to a new consumer:

cursor, claimed, _ = await r.xautoclaim("events", "workers", consumer_name,
                                        min_idle_time=60_000, count=10)
for message_id, fields in claimed:
    await handle(fields)
    await r.xack("events", "workers", message_id)

Verified: after a second of idling, a third worker claimed 10 idle messages, and their times_delivered rose from 1 to 2. The returned cursor pages through the rest, so a recovery loop continues until it comes back as 0-0.

Two details make this safe. min_idle_time must be comfortably longer than your slowest legitimate processing time, or you will steal work from a worker that is merely busy, and process it twice. And claiming should run at the top of the consume loop, before reading new messages, so a crashed peer's backlog is adopted promptly instead of after the current burst.

Verify: kill a worker mid-batch and confirm another one claims its entries within min_idle_time.

The consumer loop, including recovery 5 ordered steps. The consumer loop, including recovery XAUTOCLAIM first adopt a dead peer’s work XREADGROUP > with BLOCK wait without polling process the batch idempotently XACK the ids only what actually finished loop trim on write, not here Claiming before reading means a crashed worker’s messages are never stranded.

5. Track lag and know the limits

XINFO GROUPS reports everything worth alerting on in one call:

info = (await r.xinfo_groups("events"))[0]
# {'name': 'workers', 'consumers': 3, 'pending': 150, 'lag': 100}

lag is the number of entries never delivered to the group, pending is the number delivered but unacknowledged. They fail differently: rising lag means not enough consumers, rising pending means consumers that are failing or forgetting to ack. A dashboard with both, per group, diagnoses most incidents on sight.

Where Streams stop being the right answer is worth stating plainly. Everything lives in memory, so retention is bounded by RAM and expressed in entries rather than days. There is no partitioning, so throughput is one Redis shard's worth. And the durability is Redis's — RDB snapshots and AOF, not a replicated log. For a work queue attached to a service that already depends on Redis, that is usually the right trade; for an event history other teams will replay next year, it is not, and Kafka is.

Verify: both lag and pending are exported per group and alert separately.

Redis Streams next to a dedicated broker 2 columns contrasting Redis Streams, Kafka or RabbitMQ. Redis Streams next to a dedicated broker Redis Streams already in your stack no new service to run per-message acknowledgement retention is a MAXLEN you choose memory-bound, one shard Kafka or RabbitMQ a system of its own durable on disk by design routing or partitioned replay retention in days, not entries scales past one machine Streams are an excellent fit until retention or throughput outgrows a Redis instance.

Verification

A Streams consumer is production-ready when:

  • Every XADD trims with MAXLEN ~, and XLEN is stable.
  • Consumer names are stable per worker, not random per restart.
  • Every processed message is acknowledged, with pending returning to zero when idle.
  • Abandoned entries are claimed by XAUTOCLAIM at the top of the loop.
  • Repeatedly redelivered messages are dead-lettered based on times_delivered.
  • Lag and pending are both monitored, per group.

Pitfalls & edge cases

  • Creating the group with $. Everything already in the stream is skipped, silently.
  • Random consumer names. Pending entries are stranded under names that will never return.
  • XREADGROUP with 0 instead of >. You replay your own pending list and never see new messages.
  • min_idle_time shorter than processing time. Busy workers have their messages stolen and processed twice.
  • No trimming. The stream grows until Redis evicts or dies; MAXLEN ~ on every write is the fix.
  • Assuming Streams are durable like a log. Persistence is Redis's, and a failover can lose recent writes.

Frequently Asked Questions

How do I use Redis Streams as a work queue in asyncio?

Create a consumer group with xgroup_create(stream, group, id="0", mkstream=True), then loop on xreadgroup(group, consumer, {stream: ">"}, count=n, block=ms). Process each message and call xack. Unacknowledged messages stay in the pending entry list, where another worker can claim them.

What is the pending entry list in Redis Streams?

The record of messages delivered to a group but not yet acknowledged. It is what makes delivery reliable: if a consumer crashes, its entries remain visible and another consumer can claim them with XAUTOCLAIM. It is also a leak if your code forgets to XACK, so monitor its depth.

How do I recover messages from a crashed Redis Streams consumer?

Call xautoclaim(stream, group, new_consumer, min_idle_time=ms) at the top of your consume loop. It transfers entries idle for longer than the threshold and returns a cursor for paging. Set min_idle_time well above your slowest normal processing time so busy workers are not robbed.

How do I stop a Redis Stream from growing forever?

Pass maxlen with approximate=True on every XADD — the MAXLEN ~ form, which trims on node boundaries and is much cheaper than exact trimming. Without trimming the stream grows until Redis runs out of memory, since entries are not removed by acknowledgement.

How do I measure consumer lag for a Redis Stream?

XINFO GROUPS reports lag — entries never delivered to the group — and pending — delivered but unacknowledged — for each group. Rising lag means too few consumers; rising pending means consumers are failing or not acknowledging. Alert on them separately.