Azure Managed Redis: Pub/Sub, Streams, events, and task queues

radio_button_uncheckedshort_text1. Decouple a real-time AI processing pipeline

Picture a legal-document platform that must run OCR, entity recognition, classification, and embedding generation for hundreds of simultaneous uploads. A synchronous API would hold every request open for several seconds; ad-hoc polling and custom retries would be fragile. Event messaging lets the upload endpoint respond quickly, lets each stage scale independently, and carries status to live dashboards.

Azure Managed Redis and Azure Cache for Redis expose both Pub/Sub and Redis Streams. Pub/Sub fans a transient notification out to connected listeners. Streams retain work so coordinated consumers can process it and acknowledge completion. This chapter uses Python and redis-py, whose API evolves regularly, so validate signatures against the installed release.

• Broadcast an event to several AI services through channels and patterns.
• Build a durable task queue with Streams and consumer groups.
• Recover pending work explicitly after a consumer failure.
• Choose broadcast, coordinated distribution, or a hybrid architecture.
• Complete a Flask exercise using Microsoft Entra ID and Azure CLI.

Topic summary

Separate a fast request path from asynchronous AI processing, then choose Pub/Sub for live fan-out and Streams for retained work.

radio_button_uncheckedshort_text2. Distinguish an event, a message, and a task

A publisher should not need to know every receiver endpoint. The intermediary decouples deployment and scaling, while correlation IDs, event type, schema version, timestamp, and a reference to any large payload preserve traceability. Store documents and large model artifacts outside Redis messages and send identifiers rather than copying the content through every channel.

Topic summary

Events announce facts; tasks request work. Model the payload and delivery requirement before choosing the Redis feature.

radio_button_uncheckedshort_text3. Understand Redis Pub/Sub channel fan-out

A publisher sends PUBLISH with a channel name and payload. Redis immediately pushes the message to every active subscriber of that channel, without the publisher knowing who is listening. Each subscriber gets its own copy, so sentiment, intent, and context services can react in parallel to one new-conversation event.

This loose coupling offers low latency and high throughput across many channels. It is appropriate for real-time coordination, cache invalidation, model or embedding refresh notices, training-status changes, predictions, and operational telemetry that is useful only while listeners are online.

Topic summary

PUBLISH creates a one-to-many broadcast: every connected subscriber receives the event independently.

radio_button_uncheckedshort_text4. Design channels and pattern subscriptions

Channels are plain strings, so a predictable namespace makes ownership and filtering visible. A useful convention moves from domain to subject to event, for example ai:models:updated. Add a tenant or entity identifier only when isolation or targeted delivery needs it; uncontrolled per-user channels can become difficult to operate.

ai:conversations:42
ml:model:predictions
ai:training:status
embeddings:cache:refresh

# Pattern subscriptions
ai:conversations:*
ml:models:*:status
ai:*:predictions

SUBSCRIBE names exact channels. PSUBSCRIBE accepts glob-style patterns and delivers a pmessage containing the matched pattern, actual channel, and payload. Broad patterns simplify discovery but can increase traffic and accidentally expose unrelated events, so authorize and document namespaces.

subscription = client.pubsub()
subscription.psubscribe("ai:*")

for event in subscription.listen():
    if event["type"] == "pmessage":
        handle_ai_event(
            pattern=event["pattern"],
            channel=event["channel"],
            payload=event["data"],
        )

Topic summary

Use stable channel namespaces, exact subscriptions for narrow traffic, and carefully bounded patterns for related event families.

radio_button_uncheckedshort_text5. Account for at-most-once delivery and backpressure

• No persistence: Redis does not keep the Pub/Sub message for later replay.
• At-most-once delivery: a disconnected, restarting, or failed subscriber permanently misses the event.
• No acknowledgment: the publisher receives a subscriber count, not confirmation that business processing succeeded.
• No queue backpressure: a slow subscriber must buffer, shed load, or disconnect in its own process.
• No processing history: ordering, audit, and replay must come from another store or messaging model.

These properties are strengths when the event is replaceable or short lived. They are unacceptable when losing one document or charging twice would matter. Do not turn Pub/Sub into a work queue by giving every worker the same subscription: all workers receive the same task, multiplying compute and potentially producing conflicting writes.

Topic summary

Pub/Sub favors speed and fan-out over durability, acknowledgment, replay, and coordinated load distribution.

radio_button_uncheckedshort_text6. Apply Pub/Sub to the right AI scenarios

A subscriber failure must not block the publisher or other subscribers. If the same fact also requires a guaranteed business action, persist the work separately—often in a Stream—and use Pub/Sub only for the live view.

Topic summary

Choose Pub/Sub when all connected services should see the same replaceable event and each performs an independent reaction.

radio_button_uncheckedshort_text7. Publish and listen with redis-py

The connection below assumes TLS port 10000 for Azure Managed Redis and an already configured Microsoft Entra credential provider. In production, run the blocking listener in a dedicated worker or thread, handle reconnects, validate message schemas, and shut the subscription down cleanly.

import redis

client = redis.Redis(
    host="<cache-name>.<region>.redis.azure.net",
    port=10000,
    ssl=True,
    credential_provider=entra_provider,
    decode_responses=True,
)

client.publish("ai:models:updated", "summarizer:v3")

subscription = client.pubsub()
subscription.subscribe("ai:models:updated", "ai:embeddings:refresh")

for event in subscription.listen():
    if event["type"] == "message":
        handle_event(event["channel"], event["data"])

Inspect event["type"] because subscription confirmation frames also arrive on the connection. With decode_responses=True, channels and payloads are strings; leave decoding disabled for binary data. A publisher can use the integer returned by publish() as an observation of how many clients received the frame, but it is not a processing acknowledgment.

Topic summary

Use publish(), pubsub(), subscribe(), psubscribe(), and listen() with explicit frame handling, secure connections, and a resilient listener lifecycle.

radio_button_uncheckedshort_text8. Bridge Redis events to browser clients

Browsers normally do not connect directly to Redis. A trusted FastAPI or Flask service subscribes in the background and forwards authorized events over WebSockets. The bridge must map users to allowed topics, limit queues per socket, close stale connections, and avoid leaking a tenant event into another tenant session.

from fastapi import FastAPI, WebSocket
import redis.asyncio as redis

app = FastAPI()

async def forward_predictions(socket: WebSocket):
    client = redis.Redis(
        host="<cache-name>.<region>.redis.azure.net",
        port=10000,
        ssl=True,
        credential_provider=entra_provider,
        decode_responses=True,
    )
    async with client.pubsub() as subscription:
        await subscription.subscribe("ai:predictions:ready")
        async for event in subscription.listen():
            if event["type"] == "message":
                await socket.send_json({"type": "prediction", "data": event["data"]})

One Redis subscription per browser can exhaust connections at scale. A production gateway commonly shares one subscription per process, demultiplexes messages to local sockets, and publishes only compact status information; durable result data remains behind authenticated APIs.

Topic summary

A WebSocket gateway converts Redis fan-out into browser updates while enforcing authorization, buffering, and connection limits.

radio_button_uncheckedshort_text9. Model durable work as a Redis Stream

A Redis Stream is an append-only sequence of field-value entries. XADD appends a task and returns a time-ordered ID such as 1699980000000-0. Unlike Pub/Sub, entries remain available until the application deletes or trims them. An upload API can enqueue inference and return the ID immediately while workers process at their own pace.

task_id = client.xadd(
    "ai:inference:queue",
    {
        "user_id": "42",
        "model": "summarizer",
        "prompt": "Summarize document 917",
        "priority": "high",
    },
    maxlen=10_000,
    approximate=True,
)

# Return immediately from the API instead of waiting for inference.
return {"task_id": task_id, "status": "queued"}

Streams fit inference queues, multi-stage extract-analyze-summarize-embed pipelines, long-running jobs, processing history, and workloads that need recovery after a process crash. The entry ID gives a stable idempotency and correlation key, but it does not make the business operation exactly once.

Topic summary

XADD creates a retained, ordered work record and lets an API return before background AI processing finishes.

radio_button_uncheckedshort_text10. Distribute tasks with consumer groups

XGROUP CREATE establishes a group and can create the Stream with MKSTREAM. Workers share the group name but must use unique, case-sensitive consumer names. XREADGROUP with > requests entries never delivered to any consumer in that group. Redis assigns different new entries among active readers, so adding workers scales throughput without application-side load-balancing code.

import os
import redis

STREAM = "ai:inference:queue"
GROUP = "inference-workers"

try:
    client.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
except redis.ResponseError as error:
    if "BUSYGROUP" not in str(error):
        raise

consumer = f"{os.getenv('HOSTNAME', 'local')}-{os.getpid()}"

while True:
    batches = client.xreadgroup(
        groupname=GROUP,
        consumername=consumer,
        streams={STREAM: ">"},
        count=5,
        block=5000,
    )
    for _, tasks in batches:
        for task_id, fields in tasks:
            process_idempotently(task_id, fields)
            client.xack(STREAM, GROUP, task_id)

XACK removes the completed delivery from that group’s Pending Entries List (PEL); it does not necessarily delete the Stream entry. Reading with an ID such as 0 retrieves pending history for that consumer rather than new work. Multiple groups may consume the same Stream independently, giving each application its own coordinated copy of the sequence.

Topic summary

Consumer groups divide new entries among uniquely named workers, and XACK records successful processing for the group.

radio_button_uncheckedshort_text11. Recover failed deliveries explicitly

When a worker dies after delivery but before XACK, Redis keeps the entry in the PEL. It does not automatically reassign or retry it. Recovery code must inspect XPENDING and claim sufficiently idle work with XCLAIM or XAUTOCLAIM. A restarted consumer can also read its own pending history before requesting new entries.

# Observe entries that were delivered but not acknowledged.
pending = client.xpending_range(
    "ai:inference:queue", "inference-workers", "-", "+", count=100
)

# Redis keeps abandoned entries pending; your recovery loop must claim them.
next_id, claimed, deleted = client.xautoclaim(
    "ai:inference:queue",
    "inference-workers",
    "recovery-worker-1",
    min_idle_time=300_000,
    start_id="0-0",
    count=25,
)

for task_id, fields in claimed:
    process_idempotently(task_id, fields)
    client.xack("ai:inference:queue", "inference-workers", task_id)

Recovery creates at-least-once behavior: a timeout may cause a second worker to claim an item while the first is still finishing. Make side effects idempotent with the Stream ID or a business idempotency key, use finite retry counts, and move poison tasks to a dead-letter Stream with diagnostic context rather than retrying forever.

Topic summary

The PEL preserves unacknowledged work, but the application must claim idle entries and make repeated delivery safe.

radio_button_uncheckedshort_text12. Monitor and bound Stream retention

XINFO STREAM exposes length and IDs; XINFO GROUPS reports lag and pending counts; XINFO CONSUMERS reveals active consumers, idle time, and pending ownership. Alert on growing lag, old pending entries, excessive delivery attempts, worker churn, Redis memory, latency, and command errors.

stream = client.xinfo_stream("ai:inference:queue")
groups = client.xinfo_groups("ai:inference:queue")
consumers = client.xinfo_consumers("ai:inference:queue", "inference-workers")

client.xtrim(
    "ai:inference:queue",
    maxlen=10_000,
    approximate=True,
)

• Trim by an evidence-based maximum length with XADD MAXLEN or XTRIM; approximate trimming is cheaper.
• Keep payloads small and place documents or embeddings in durable storage.
• Choose a retention window long enough for audit and recovery, but do not assume an unbounded Stream expires automatically.
• Remember that persistence and acknowledgment add some latency and code compared with Pub/Sub.
• Test consumer shutdown, reconnection, poison messages, and recovery under realistic load.

Topic summary

Observe Stream, group, and consumer state, then cap retained entries so reliability does not become uncontrolled memory growth.

radio_button_uncheckedshort_text13. Choose broadcast or coordinated distribution

Use Pub/Sub for cache invalidation, configuration refresh, metrics, and disposable notifications. Use Streams for jobs, inference requests, pipelines, and work that must remain recoverable. If strict transactions, advanced routing, dead-lettering, or cross-system delivery guarantees dominate, compare a purpose-built Azure messaging service rather than forcing every workload into Redis.

Topic summary

Broadcast facts with Pub/Sub; coordinate durable work with Streams; evaluate a dedicated broker when requirements exceed either model.

radio_button_uncheckedshort_text14. Combine Streams and Pub/Sub in one architecture

Many AI systems need both semantics. When a document arrives, add the durable processing request to a Stream and publish a short-lived received event. One member of the worker group performs each delivery, while WebSocket, analytics, and monitoring subscribers all see status broadcasts. Later pipeline stages can append to new Streams and publish progress events.

import json

# Durable work: one worker in the group handles each delivery.
task_id = client.xadd("ai:documents:queue", {
    "document_id": "doc-917",
    "requested_by": "user-42",
})

# Ephemeral fan-out: every connected observer sees the status event.
client.publish("ai:documents:events", json.dumps({
    "event": "document_received",
    "task_id": task_id,
}))

The two writes are not one atomic business transaction by default. If losing the notification is acceptable, treat it as best effort. If they must remain consistent, use an outbox or another durable event record and publish from that record. Include the same correlation ID in Stream entries, Pub/Sub events, logs, and result storage.

Topic summary

A hybrid design uses Streams for the authoritative job and Pub/Sub for ephemeral observability, tied together by correlation and explicit consistency rules.

radio_button_uncheckedshort_text15. Guided lab: publish and subscribe in Azure Managed Redis

The source exercise estimates 30–40 minutes. It builds a Python Flask page that publishes and subscribes in real time from one interface. Use a disposable resource, do not commit credentials, and remove the environment when the verification is complete.

1. Prepare an Azure subscription, Visual Studio Code, Python 3.12 or later, the latest Azure CLI, and the redisenterprise extension installed with az extension add --name redisenterprise.
2. Download the starter project, create an isolated virtual environment, and install its pinned dependencies.
3. Create an Azure Managed Redis resource and grant the development identity the required data access.
4. Connect through TLS with Microsoft Entra ID; keep credentials in the normal Azure credential chain.
5. Complete helpers that publish an event, publish to every configured channel, and format received frames.
6. Run the listener on a background thread so the Flask request loop remains responsive.
7. Subscribe to exact channels and patterns, run the page, and verify that messages appear live.
8. Test a temporary disconnect to observe that Pub/Sub does not replay missed events, then clean up the Azure resource.

Topic summary

The lab validates secure provisioning, background listening, channel and pattern subscriptions, live publishing, delivery semantics, and cleanup.

radio_button_uncheckedshort_text16. Assessment review and production checklist

1. Pub/Sub serves active subscribers only; Streams retain entries for later or coordinated consumption.
2. Choose Streams and consumer groups when a pipeline needs acknowledgment and application-managed retry.
3. XADD appends an entry to a Stream.
4. Pub/Sub best fits real-time status broadcast to currently connected clients or services.
5. XREADGROUP with > distributes new entries among consumers in one group; idempotency protects against repeat delivery.

• Version message schemas and include correlation and idempotency identifiers.
• Authorize channel namespaces and keep large payloads in external storage.
• Give every Stream worker a unique consumer name.
• Acknowledge only after the business side effect succeeds.
• Monitor PEL age, lag, retry count, memory, latency, and disconnected subscribers.
• Implement XAUTOCLAIM or XPENDING plus XCLAIM; Redis does not recover abandoned work by itself.
• Trim Streams and route poison tasks to a dead-letter strategy.
• Use Microsoft Entra ID, TLS, least privilege, and tested reconnect behavior.

Official references

1. Azure Managed Redis overview
2. Create a Python app with Azure Managed Redis and Microsoft Entra ID
3. Use Microsoft Entra ID for Azure Managed Redis authentication
4. Publisher-Subscriber pattern — Azure Architecture Center
5. Background jobs guidance — Azure Architecture Center
6. Redis Pub/Sub reference
7. Redis Streams and consumer groups reference

Topic summary

Production messaging joins the correct distribution model with secure identity, explicit recovery, idempotency, bounded retention, and observability.