How to Read Redis Streams with XREADGROUP in Python

Read new messages from a Redis stream using a consumer group with XREADGROUP, handling JSON payloads and group creation.

Medium Python 3.9+ Aug 9, 2026 Streaming & messaging 12 views 0 copies

redis streams consumer groups messaging xreadgroup

Requires third-party packages — install first

pip install redis

Python code

40 lines

Python 3.9+

import redis
import json

def read_group_messages(stream_key, group_name, consumer_name, count=10):
    r = redis.Redis(host="localhost", port=6379, decode_responses=True)
    try:
        r.xgroup_create(stream_key, group_name, id="0", mkstream=True)
    except redis.exceptions.ResponseError:
        pass

    messages = r.xreadgroup(
        group_name,
        consumer_name,
        {stream_key: ">"},
        count=count,
        block=2000
    )

    if not messages:
        return []

    parsed = []
    for _, entries in messages:
        for message_id, fields in entries:
            parsed.append({
                "id": message_id,
                "data": json.loads(fields.get("data", "{}"))
            })
    return parsed

if __name__ == "__main__":
    # Simulate a producer adding messages first
    r = redis.Redis(host="localhost", port=6379, decode_responses=True)
    for i in range(3):
        payload = json.dumps({"event": f"item_{i}", "value": i * 10})
        r.xadd("events", {"data": payload})

    result = read_group_messages("events", "workers", "consumer-1", count=5)
    for msg in result:
        print(f"Message {msg['id']}: {msg['data']}")

Output

stdout

Message 1720000000000-0: {'event': 'item_0', 'value': 0}
Message 1720000000000-1: {'event': 'item_1', 'value': 10}
Message 1720000000000-2: {'event': 'item_2', 'value': 20}

How it works

The xgroup_create call ensures the consumer group exists, using mkstream=True to create the stream if needed, and catching ResponseError to ignore duplicates. The xreadgroup command reads new messages (specified by ">") for the consumer without affecting other group members. Each message is returned as a list of tuples containing stream key and entries, which are parsed to extract the message ID and JSON data. The decoded responses from Redis make the fields naturally accessible as Python dictionaries.

Common mistakes

Forgetting `decode_responses=True` so you get bytes needing manual decoding.
Not catching `ResponseError` when the group already exists, causing the script to fail on rerun.
Assuming `xreadgroup` blocks forever without specifying `block` — it can return `None` if timeout is reached.
Mixing up `xreadgroup` (consumer group) with `xread` (plain read).

Variations

Use `count=-1` to read all pending messages for the consumer after a crash and reprocess them.
Read historical messages with `id='0'` instead of `">"` to get already-delivered entries.

Real-world use cases

Distributing job tasks to multiple workers where each task is consumed once by one worker in a queue-like pattern.
Building a real-time event processing pipeline that ingests user actions from a web application and aggregates analytics.
Implementing a log aggregation service that consumes log entries from a central stream and forwards them to storage or sinks.

How to Read Redis Streams with XREADGROUP in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes