Redis Streams in Python: Producers and Consumers#

Use Redis Streams in Python to append messages, read them asynchronously and distribute work across a consumer group. FastStream uses redis-py for Redis access and adds subscriber and publisher decorators, message validation and in-memory testing. The example below publishes a message and processes it with a stream subscriber.

Redis Streams are append-only logs introduced in Redis 5.0. Each entry has an ID and a set of field-value pairs. Unlike Pub/Sub, entries remain available for later reads until they are deleted or trimmed; durability across server restarts depends on your Redis persistence configuration.

A stream key is stored on one Redis node. Consumer groups distribute entries across workers, but do not partition one stream into Kafka-style partitions. Completion order can differ from stream order when workers process messages concurrently.

Run a Python producer and consumer#

Install the Redis backend and CLI:

pip install "faststream[redis,cli]"

With Redis listening on localhost:6379, save this application as redis_streams.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from faststream import FastStream, Logger
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber(stream=StreamSub("test-stream", group="test-group", consumer="1"))
async def handle(msg: str, logger: Logger):
    logger.info(msg)


@app.after_startup
async def t():
    await broker.publish("Hi!", stream="test-stream")

Start it with:

faststream run redis_streams:app

After startup, the application publishes "Hi!" to test-stream. The subscriber reads it as part of test-group and logs the message. FastStream acknowledges successfully processed messages by default. To publish another message from Python, use await broker.publish("Hi!", stream="test-stream") while the broker is connected.

Choose how to read and recover messages#

• Stream subscriptions: read entries with XREAD, or configure XREADGROUP for a group.
• Consumer groups: share work across workers with distinct consumer names.
• Acknowledgements: understand XACK and control when processing is confirmed.
• Pending-message recovery: use XAUTOCLAIM to recover work abandoned by a worker.
• Publishing: append messages with XADD and configure stream length.