Redis Streams

Redis Streams, introduced in Redis 5.0, is a data structure suitable for lightweight, reliable message queue scenarios.

Installation

pip install onestep-redis

Quick Start

Start Redis

Quick start with Docker:

docker run -d --name redis \
  --restart=always \
  -p 6379:6379 \
  redis:7

Basic Usage

from onestep import OneStepApp
from onestep_redis import RedisConnector

app = OneStepApp("redis-demo")

# Create connection
redis = RedisConnector("redis://localhost:6379")

# Create Stream as Source
source = redis.stream(
    "jobs",
    group="workers",
    batch_size=100,
    poll_interval_s=0.5,
)

# Create Stream as Sink
sink = redis.stream("processed")


@app.task(source=source, emit=sink, concurrency=8)
async def process_job(ctx, item):
    print(f"Processing job: {item}")
    return {"job": item["job"], "status": "done"}


if __name__ == "__main__":
    app.run()

Stream Configuration

Basic Parameters

source = redis.stream(
    "my_stream",            # Stream name
    group="my_group",       # Consumer group name
    consumer=None,          # Consumer name (auto-generated by default)
    batch_size=100,         # Messages fetched per poll
    poll_interval_s=0.5,    # Poll interval (seconds)
)

Consumer Groups

Consumer groups allow multiple consumers to share the same Stream:

# Consumer group is auto-created (if it doesn't exist)
source = redis.stream(
    "jobs",
    group="workers",
)

# Multiple processes using the same group name
# Redis automatically distributes messages to different consumers

Stream Trimming

Control Stream size:

source = redis.stream(
    "jobs",
    group="workers",
    maxlen=10000,  # Keep the latest 10000 messages
)

Publishing Messages

Publish via Sink

Task return values are automatically published:

@app.task(source=..., emit=sink)
async def process(ctx, item):
    return {"result": "data"}  # Auto-published to sink

Manual Publish

import asyncio

async def main():
    sink = redis.stream("my_stream")
    
    # Publish single
    await sink.publish({"job": "data"})
    
    # Publish multiple
    for i in range(100):
        await sink.publish({"id": i})

asyncio.run(main())

Acknowledgment Mechanism

Redis Streams messages are automatically acknowledged (XACK) after successful task completion:

• Success: auto XACK
• Retry: message stays in PEL (Pending Entries List), next poll reads pending messages first
• Fail: message is XACK'd from PEL; if dead_letter is configured, it's written to the dead letter Sink first

Multiple Consumers

Multiple consumers can be started on the same Stream for load balancing:

# Run the same code on multiple machines
# Redis Streams automatically distributes messages
@app.task(source=source, concurrency=4)
async def process(ctx, item):
    ...

YAML Configuration

resources:
  redis:
    type: redis
    url: "redis://localhost:6379"
  
  jobs:
    type: redis_stream
    connector: redis
    stream: "jobs"
    group: "workers"
    batch_size: 100
  
  results:
    type: redis_stream
    connector: redis
    stream: "results"

tasks:
  - name: process_jobs
    source: jobs
    emit: results
    concurrency: 8

Comparison with RabbitMQ

Feature	Redis Streams	RabbitMQ
Message Persistence	✅ AOF/RDB	✅ Durable queues
Consumer Groups	✅ Native	✅ Configurable
Message Acknowledgment	✅ XACK	✅ Ack/Nack
Dead Letter Queue	❌ Manual	✅ Native
Delayed Messages	❌ Extra work	✅ Plugin
Throughput	Very high	High
Deployment Complexity	Low	Medium
Management UI	Redis CLI / Insight	Web UI

Selection Advice:

• Redis Streams: Lightweight, high throughput, existing Redis infrastructure
• RabbitMQ: Complex routing, delayed messages, dead letter queues needed

Best Practices

1. Batch Size

Adjust based on task processing time and memory:

# High throughput: larger batch_size
source = redis.stream("high_volume", group="workers", batch_size=500)

# Low latency: smaller batch_size
source = redis.stream("low_latency", group="workers", batch_size=10)

2. Poll Interval

# High throughput: shorter interval
source = redis.stream(..., poll_interval_s=0.1)

# Low load: longer interval (saves CPU)
source = redis.stream(..., poll_interval_s=1.0)

3. Stream Naming

Use meaningful naming conventions:

# Business-focused naming
redis.stream("orders:created", group="order-processor")
redis.stream("payments:pending", group="payment-worker")

# Environment isolation
redis.stream(f"prod:jobs", group="workers")
redis.stream(f"dev:jobs", group="workers")

4. Monitor Stream Length

# Check Stream status on application startup
async def check_stream_health():
    info = await redis.client.xinfo_stream("jobs")
    length = info["length"]
    if length > 100000:
        print(f"Warning: Stream is too long ({length} messages)")

5. Graceful Shutdown

from onestep import OneStepApp
from onestep_redis import RedisConnector

app = OneStepApp("redis-demo", shutdown_timeout_s=30.0)

# shutdown_timeout_s controls the time to wait for inflight tasks to complete on shutdown