Context & Why Redis Streams
After working with MySQL, PostgreSQL, and MongoDB across different projects, I kept running into the same problem: databases are great at storing state, but terrible at passing events between services. When we needed real-time order processing, user activity tracking, or cross-service notifications, we’d bolt on polling loops or simple task queues — and it always felt like the wrong tool for the job.
Redis Streams, introduced in Redis 5.0, changed that. Basic pub/sub drops messages if a subscriber is offline. Simple lists lack consumer group semantics. Streams give you a persistent, append-only log with acknowledgment and delivery tracking built in — none of those compromises.
Here’s when Redis Streams makes more sense than Kafka or RabbitMQ:
- You’re already running Redis and want to avoid another infrastructure dependency
- Your message volume is moderate (millions per day, not billions)
- You need fast setup with low operational overhead
- You want consumer group semantics without Kafka’s partition management complexity
Installation
Installing Redis 7+ with Streams Support
Redis Streams have been stable since Redis 5.0, but Redis 7 brought important improvements to XAUTOCLAIM and consumer group management. I recommend Redis 7.x for any new setup.
On Ubuntu/Debian:
# Add the official Redis repository
curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list
sudo apt update
sudo apt install -y redis-server
# Verify version — should be 7.x
redis-server --versionOn CentOS/RHEL:
sudo dnf install -y epel-release
sudo dnf install -y redis
sudo systemctl enable --now redisVerify Streams commands are available:
redis-cli XADD test_stream '*' key value
# Should return something like: "1728390000000-0"
redis-cli DEL test_streamPython Client Setup
pip install "redis>=5.0.0"Configuration
Adding Messages with XADD
XADD appends a message to a stream. Each message gets an auto-generated ID — millisecond timestamp plus a sequence number — unless you specify one explicitly.
# Basic XADD: auto-generate ID
XADD orders '*' user_id 42 product_id 1001 quantity 2 status pending
# With MAXLEN to cap stream size (approximate trimming with ~)
XADD orders MAXLEN ~ 10000 '*' user_id 43 product_id 1002 quantity 1 status pendingI always add MAXLEN ~ 10000 in production. Without it, streams grow indefinitely and Redis eventually runs out of memory. The ~ tells Redis to trim lazily — only when it makes structural sense — which is far cheaper than exact trimming on every single write.
In Python:
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# Add an order event
msg_id = r.xadd(
'orders',
{
'user_id': '42',
'product_id': '1001',
'quantity': '2',
'status': 'pending'
},
maxlen=10000,
approximate=True
)
print(f"Message added: {msg_id}")Creating Consumer Groups
Consumer groups let multiple workers read from the same stream without duplicates. Each message goes to exactly one consumer in the group — not broadcast to all of them.
# Create a group starting from the beginning of stream
# 0 = read all existing messages
# $ = only new messages from now on
# MKSTREAM creates the stream if it doesn't exist
XGROUP CREATE orders order-processors 0 MKSTREAM# Create consumer group, skip if already exists
try:
r.xgroup_create('orders', 'order-processors', id='0', mkstream=True)
print("Consumer group created")
except redis.exceptions.ResponseError as e:
if 'BUSYGROUP' in str(e):
print("Group already exists, skipping")
else:
raiseConsuming Messages with XREADGROUP
XREADGROUP reads new messages and moves them into the consumer’s Pending Entry List (PEL). Messages stay in the PEL until you call XACK — that’s your at-least-once delivery guarantee.
# Read up to 10 new messages for consumer "worker-1"
# The ">" means "give me new undelivered messages"
XREADGROUP GROUP order-processors worker-1 COUNT 10 BLOCK 2000 STREAMS orders >Swap > for 0 and you re-read pending (unacknowledged) messages — handy for recovering after a crash.
import time
import socket
consumer_name = f"worker-{socket.gethostname()}"
def process_orders():
while True:
# Read new messages, block up to 2 seconds if stream is empty
messages = r.xreadgroup(
groupname='order-processors',
consumername=consumer_name,
streams={'orders': '>'},
count=10,
block=2000
)
if not messages:
continue
for stream_name, stream_messages in messages:
for msg_id, fields in stream_messages:
try:
process_single_order(msg_id, fields)
# Acknowledge: removes from PEL
r.xack('orders', 'order-processors', msg_id)
except Exception as e:
print(f"Failed to process {msg_id}: {e}")
# Don't ack — message stays in PEL for retry
def process_single_order(msg_id, fields):
print(f"Processing order {msg_id}: {fields}")
time.sleep(0.1) # Simulate workDead Letter Queue with XPENDING and XCLAIM
Most guides cover XREADGROUP and stop there. The reaper pattern is what actually makes Streams reliable in production. When a consumer crashes mid-processing, its messages stay stuck in the PEL until something comes along to claim them. You need a dedicated reaper process: scan for idle PEL entries, retry them, and after N failures, move them to a dead letter queue (DLQ).
# Inspect pending messages (all consumers, all IDs)
XPENDING orders order-processors - + 100DLQ_STREAM = 'orders:dead-letter'
MAX_RETRIES = 3
CLAIM_IDLE_MS = 5 * 60 * 1000 # 5 minutes
def reaper_loop():
"""Reclaim stale messages and route persistent failures to DLQ."""
while True:
# Find messages idle longer than CLAIM_IDLE_MS
pending = r.xpending_range(
'orders',
'order-processors',
min='-',
max='+',
count=50,
idle=CLAIM_IDLE_MS
)
for entry in pending:
msg_id = entry['message_id']
delivery_count = entry['times_delivered']
if delivery_count >= MAX_RETRIES:
# Fetch message content before removing from PEL
msgs = r.xrange('orders', min=msg_id, max=msg_id, count=1)
if msgs:
_, fields = msgs[0]
fields['original_id'] = msg_id
fields['failure_reason'] = f'exceeded {MAX_RETRIES} retries'
# Write to DLQ
r.xadd(DLQ_STREAM, fields, maxlen=5000, approximate=True)
# Ack to remove from PEL — stop retrying
r.xack('orders', 'order-processors', msg_id)
print(f"Moved {msg_id} to DLQ after {delivery_count} attempts")
else:
# Reassign to this reaper consumer for retry
r.xclaim(
'orders',
'order-processors',
consumer_name,
min_idle_time=CLAIM_IDLE_MS,
message_ids=[msg_id]
)
print(f"Reclaimed {msg_id} for retry (attempt {delivery_count + 1})")
time.sleep(30) # Run every 30 secondsOne thing I learned the hard way: always set MAXLEN on your DLQ too. A bug that keeps producing failures will silently fill memory if the DLQ grows unchecked.
If you’re on Redis 6.2+, XAUTOCLAIM combines the scan and claim into a single atomic command — cleaner than the two-step XPENDING + XCLAIM pattern above. Worth switching to on fresh setups.
Verification & Monitoring
Checking Stream and Group Health
# Stream length
XLEN orders
# Full stream info (first/last entry, groups count, length)
XINFO STREAM orders
# Consumer group details (pending count, last delivered ID)
XINFO GROUPS orders
# Per-consumer PEL size
XINFO CONSUMERS orders order-processorsHealth Check Script
Watch the pending count per consumer. If it keeps climbing, your consumer is either too slow or silently stuck — both problems you want to catch early, not after a backlog of 50,000 unprocessed messages.
def stream_health_check():
groups = r.xinfo_groups('orders')
for group in groups:
name = group['name']
pending = group['pending']
last_delivered = group['last-delivered-id']
print(f"Group: {name}")
print(f" Pending messages: {pending}")
print(f" Last delivered ID: {last_delivered}")
if pending > 1000:
print(f" WARNING: High pending count — consumers may be falling behind")
# Check DLQ
dlq_len = r.xlen(DLQ_STREAM) if r.exists(DLQ_STREAM) else 0
print(f"\nDead Letter Queue size: {dlq_len}")
if dlq_len > 0:
print(" Action required — review failed messages in DLQ")
stream_health_check()Enable Redis Persistence
Redis is in-memory by default. For any queue where losing messages matters, enable persistence in /etc/redis/redis.conf:
# AOF: log every write command (best durability)
appendonly yes
appendfsync everysec
# Or RDB snapshots (lighter, some data loss risk)
save 900 1
save 300 10
save 60 10000Then restart Redis: sudo systemctl restart redis
Quick Tips from Running This in Production
- Always set MAXLEN with
~— exact trimming on every XADD is expensive; approximate trimming is practically identical in behavior - Use hostname-based consumer names — makes it easy to trace which pod owns which stale PEL entries
- Run the reaper as a separate process — mixing it into your main consumer loop adds unnecessary complexity and failure modes
- Set idle timeout to 2–3× your expected max processing time — too short triggers false reclaims; too long delays recovery after crashes
- Monitor PEL size per consumer — export
XINFO CONSUMERSdata to Prometheus or Grafana if you’re running at scale - Always MAXLEN your DLQ — a runaway failure loop will fill memory if you don’t cap it
Redis Streams won’t replace Kafka at millions of messages per second. For most teams already running Redis, though, it removes a whole layer of infrastructure. The consumer group model is solid. The commands are straightforward. Wire up the reaper and you have genuine at-least-once delivery guarantees — no separate broker cluster needed.
