Pub/Sub

Overview

Pub/Sub is one of Redis's simplest and most useful features. The idea is direct: a process PUBLISHES a message on a "channel"; any process SUBSCRIBED to that channel receives it immediately. There's no complex configuration, no additional infrastructure, no new services — it all runs inside the Redis you already have working.

You'll use it for two main cases: cache invalidation events between components (when a value changes, every service invalidates its related caches) and lightweight notifications (when something happens, broadcast the event to UI WebSockets). It's the perfect complement to module 2's caching: capsule 04 showed 3 invalidation strategies; here you implement the "event-driven" one for real, not just conceptually.

But this capsule is honest about Pub/Sub's critical limitations. It isn't a message broker. It has no persistence, no acknowledgment, no replay. If you publish and nobody's listening, the message is lost forever. If your subscriber crashes for 10 seconds, the messages from those 10 seconds are gone. For cases where you CAN'T lose messages (transactions, legal auditing), Pub/Sub is NOT the tool — you need RabbitMQ, Kafka, or Redis Streams (out of scope). You'll leave the capsule knowing when Pub/Sub is perfect and when it's inadequate.

Pub/Sub: the basic commands

The model

Publisher                    Subscriber 1     Subscriber 2
   │                              │                │
   │ PUBLISH news "hello"         │                │
   ├──────────────────────────────┼────────────────┤
   │                              │                │
   │                      receives "hello"  receives "hello"

3 commands make up Pub/Sub:

• PUBLISH channel message — publishes a message on the channel
• SUBSCRIBE channel [channel ...] — listens to one or more channels
• PSUBSCRIBE pattern [pattern ...] — listens to channels matching a pattern (with the wildcards *, ?)

A demo in redis-cli

You need TWO terminals open with redis-cli.

Terminal 1 (the subscriber):

redis-cli

127.0.0.1:6379> SUBSCRIBE news
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "news"
3) (integer) 1

The subscriber is now waiting for messages on the news channel.

Terminal 2 (the publisher):

redis-cli

127.0.0.1:6379> PUBLISH news "Hello, world"
(integer) 1   # 1 = one subscriber received the message

Terminal 1 (the subscriber receives it):

1) "message"
2) "news"
3) "Hello, world"

The subscriber got the message in real time. The latency is < 1 ms.

Multiple subscribers

Open a third terminal and subscribe to the same channel:

Terminal 3:

127.0.0.1:6379> SUBSCRIBE news

Now publish another message from terminal 2:

127.0.0.1:6379> PUBLISH news "Second message"
(integer) 2   # ✓ now there are 2 subscribers, and both receive it

Terminals 1 and 3 receive the message simultaneously.

Multiple channels

One publisher can send to different channels. One subscriber can listen to several.

A multi-channel subscriber:

127.0.0.1:6379> SUBSCRIBE news sports tech

The publisher:

127.0.0.1:6379> PUBLISH sports "Goal!"
127.0.0.1:6379> PUBLISH tech "Python 4 released"

The subscriber receives both.

Pattern subscriptions with PSUBSCRIBE

Sometimes you don't know the exact channel names — you want to subscribe to every one matching a pattern.

127.0.0.1:6379> PSUBSCRIBE cache:invalidate:*

This subscribes to:

• cache:invalidate:product:42
• cache:invalidate:user:1
• cache:invalidate:category:5

And to any other cache:invalidate:<whatever>.

The supported wildcards:

• * — any sequence of characters
• ? — exactly one character
• [abc] — one of the listed characters

Examples:

PSUBSCRIBE user:*:notifications     # user:1:notifications, user:42:notifications, etc.
PSUBSCRIBE *:invalidate             # cache:invalidate, session:invalidate, etc.
PSUBSCRIBE event.[ABC]              # event.A, event.B, event.C

When to use SUBSCRIBE vs PSUBSCRIBE

• SUBSCRIBE: you know the names of the channels you care about
• PSUBSCRIBE: the list of channels is dynamic or has a hierarchical structure

Pub/Sub from Python with redis.asyncio

Let's move to the async client (which is what we'll use in production).

An async publisher

import asyncio
from redis.asyncio import Redis


async def publish_demo():
    r = Redis(host='localhost', port=6379, decode_responses=True)

    # Publish 5 messages
    for i in range(5):
        subscribers = await r.publish("news", f"Message #{i}")
        print(f"Published {i}: {subscribers} subscribers received it")
        await asyncio.sleep(0.5)

    await r.close()


if __name__ == "__main__":
    asyncio.run(publish_demo())

publish() returns an int: how many subscribers received the message. If it returns 0, the message was lost (nobody was listening).

An async subscriber

import asyncio
from redis.asyncio import Redis


async def subscribe_demo():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()

    # Subscribe to the channel
    await pubsub.subscribe("news")
    print("Subscribed to 'news'. Waiting for messages...")

    try:
        async for message in pubsub.listen():
            # message is a dict
            if message["type"] == "message":
                print(f"Received: {message['data']}")
    finally:
        await pubsub.unsubscribe("news")
        await pubsub.close()
        await r.close()


if __name__ == "__main__":
    asyncio.run(subscribe_demo())

Testing the publisher + subscriber

Open two terminals:

Terminal 1:

python subscriber.py
# Subscribed to 'news'. Waiting for messages...

Terminal 2:

python publisher.py
# Published 0: 1 subscribers received it
# Published 1: 1 subscribers received it
# ...

Terminal 1, updated:

Subscribed to 'news'. Waiting for messages...
Received: Message #0
Received: Message #1
Received: Message #2
...

The message's structure in Python

When you receive a message, it comes as a dict:

{
    "type": "message",     # "subscribe", "unsubscribe", "message", "pmessage", "psubscribe", "punsubscribe"
    "pattern": None,       # only if it came from a PSUBSCRIBE, it holds the pattern
    "channel": "news",     # the original channel
    "data": "Hello"        # the message (a string if decode_responses=True)
}

Filter by type == "message" to ignore the confirmation messages from subscribe/unsubscribe.

A pattern subscription in Python

async def psubscribe_demo():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()

    await pubsub.psubscribe("cache:invalidate:*")
    print("Subscribed to 'cache:invalidate:*'")

    async for message in pubsub.listen():
        if message["type"] == "pmessage":
            channel = message["channel"]
            pattern = message["pattern"]
            data = message["data"]
            print(f"[{pattern}] {channel}: {data}")

Note: use pmessage (not message) for messages matching patterns.

Use case 1: Event-driven cache invalidation

Here's the fundamental use case for Pub/Sub in the backend.

The scenario

Your API has 3 services:

• The API service: it serves GET /products with cache-aside
• The admin service: from an admin panel, admins edit products
• The reports service: it generates cached reports with product data

When the admin edits a product, all 3 services should invalidate their related caches — but they don't want to be coupled by calling each other.

The solution with Pub/Sub

# admin_service.py (the publisher)
import asyncio
import json
from redis.asyncio import Redis


r = Redis(host='localhost', port=6379, decode_responses=True)


async def update_product(product_id: int, data: dict):
    # 1. Update the DB
    db.products.update(product_id, data)

    # 2. Publish the event (the services react)
    await r.publish(
        "cache:invalidate:product",
        json.dumps({
            "product_id": product_id,
            "fields": list(data.keys()),
            "timestamp": time.time(),
        })
    )

    return {"updated": True}

# api_service.py (a subscriber)
import asyncio
import json
from redis.asyncio import Redis


async def listen_for_invalidations():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()

    await pubsub.subscribe("cache:invalidate:product")
    print("API service: listening for product invalidations...")

    async for message in pubsub.listen():
        if message["type"] != "message":
            continue

        event = json.loads(message["data"])
        product_id = event["product_id"]

        # Invalidate the local caches
        await r.delete(f"cache:product:{product_id}")
        await r.delete("cache:products:list")

        print(f"  Invalidated the cache for product:{product_id}")

# reports_service.py (another subscriber)
async def listen_for_invalidations():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()

    await pubsub.subscribe("cache:invalidate:product")
    print("Reports service: listening...")

    async for message in pubsub.listen():
        if message["type"] != "message":
            continue

        event = json.loads(message["data"])
        product_id = event["product_id"]

        # Invalidate the reports that include this product
        await r.delete(f"report:product:{product_id}:weekly")
        await r.delete(f"report:product:{product_id}:monthly")
        # ...

The complete flow

1. The admin edits product:42
        │
        ▼
2. admin_service:
   - UPDATE the DB
   - PUBLISH "cache:invalidate:product" {product_id: 42, ...}
        │
        ▼
3. Redis distributes the message
        │
        ├──────► api_service: invalidates cache:product:42
        ├──────► reports_service: invalidates report:product:42:*
        └──────► (future) other_service: ...

Why this is elegant

• Decoupled: admin_service doesn't know who's listening. If you add a new service, you just add a subscription to it
• Scalable: N subscribers have constant latency (Redis distributes in parallel)
• Simple: one PUBLISH + one SUBSCRIBE in each service. Zero additional infrastructure

The pattern with hierarchical channels

For more granularity:

# The admin publishes on a specific channel
await r.publish(f"cache:invalidate:product:{product_id}", json.dumps({...}))

# The subscribers use PSUBSCRIBE
await pubsub.psubscribe("cache:invalidate:product:*")
# It receives: cache:invalidate:product:42, cache:invalidate:product:99, etc.

Useful when you want finer filters (e.g., one service only cares about certain kinds of products).

Use case 2: Real-time notifications via WebSocket

User A is editing a document
        │
        ▼
api_service: PUBLISH "doc:42:changes" {...}
        │
        ▼
The WebSocket bridge (a subscriber):
   - Receives the message
   - Broadcasts it to the users connected to doc:42 via WebSocket
        │
        ▼
Users B and C see the change in real time

A basic implementation

# The WebSocket bridge service
import asyncio
from contextlib import asynccontextmanager

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


active_connections = {}  # doc_id → a list of WebSockets


async def redis_subscriber():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()

    await pubsub.psubscribe("doc:*:changes")

    async for message in pubsub.listen():
        if message["type"] != "pmessage":
            continue

        # Extract the doc_id from the channel
        channel = message["channel"]  # "doc:42:changes"
        doc_id = channel.split(":")[1]

        # Broadcast to every WebSocket connected to that doc
        if doc_id in active_connections:
            for ws in active_connections[doc_id]:
                try:
                    await ws.send_json({"type": "doc_change", "data": message["data"]})
                except Exception:
                    pass  # the client disconnected, ignore it


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(redis_subscriber())
    yield
    task.cancel()


app = FastAPI(lifespan=lifespan)


@app.websocket("/ws/{doc_id}")
async def websocket_endpoint(websocket: WebSocket, doc_id: str):
    await websocket.accept()

    active_connections.setdefault(doc_id, []).append(websocket)

    try:
        while True:
            # Keep the connection open
            await websocket.receive_text()
    except Exception:
        pass
    finally:
        active_connections[doc_id].remove(websocket)

This pattern is used in chat apps, collaborative editing (Notion, Google Docs), and live dashboards.

Critical limitations (the "fine print")

Here's where Pub/Sub gets honest. These points are critical so you don't use Pub/Sub in the wrong scenarios.

Limitation 1: Fire-and-forget (no persistence)

If you publish and nobody's listening, the message is lost forever.

# The subscriber is NOT running
await r.publish("important_event", "critical data")
# This returns 0 (zero subscribers)
# The message vanished. There's no way to recover it.

The implication: if you need a delivery guarantee (auditing, transactions), Pub/Sub is NOT the tool.

Limitation 2: No replay

Even if your subscriber was running when the message was published, if you later add a NEW subscriber, it does NOT see the previous messages.

# T=0: the publisher publishes
# T=1: a new subscriber starts up
# The subscriber does NOT receive the message from T=0. It's as if it never existed.

The implication: Pub/Sub is no good for "replaying events" or "processing from the beginning."

Limitation 3: No acknowledgment

The publisher doesn't know whether the subscribers processed the message successfully.

subscribers = await r.publish("event", "data")
# subscribers = 3 (3 received it)
# But you do NOT know whether all 3 processed it correctly.
# One of them could have crashed while processing it.

The implication: no retry mechanism. If processing fails, there's no way to know.

Limitation 4: Downed subscribers lose everything

If your subscriber is down for 10 seconds, every message published during those 10 seconds is permanently lost.

T=0: the subscriber crashes
T=1 to T=10: the publisher sends 100 messages (all lost for this subscriber)
T=11: the subscriber restarts
T=12+: the subscriber starts receiving only the NEW messages

The implication: Pub/Sub assumes subscribers are always available. That isn't realistic for real production systems.

Limitation 5: Distribution to ALL the subscribers

When you publish, every subscriber receives the message. There's no queueing where "one of N workers processes each message."

# 3 workers subscribed to the "jobs" channel
# PUBLISH job_data → all 3 receive the same message
# All 3 process the same job → duplicated work

The implication: Pub/Sub is NOT a task queue. For load balancing jobs, use Redis Streams or a real broker.

When Pub/Sub is the right tool

✅ Pub/Sub is perfect for:

1. Cache invalidation events

• If the subscriber is down, it isn't critical — the cache eventually expires with its TTL
• Idempotent: invalidating twice is the same as invalidating once

2. Real-time notifications

• A lost notification = a UX annoyance, not critical
• If the user isn't connected to the WebSocket, the event doesn't matter to them

3. Broadcasting configuration

• Notifying workers that the config changed
• The workers would have to reload the config on their next sync anyway

4. Distributed logs/metrics (lightweight)

• If you lose some logs, it isn't critical
• For production-grade logs, use Loki/CloudWatch/etc.

❌ Pub/Sub is NOT the tool for:

1. Financial transactions

• A lost message = lost money
• You need RabbitMQ with persistence + acknowledgment

2. Order processing

• Each order has to be processed exactly once
• Pub/Sub broadcasts to everyone = duplicated processing

3. Legal auditing

• A lost log = a legal problem
• You need durable storage (a DB + a write-ahead log)

4. Job queues

• A job should be run by ONE worker, not all of them
• Use Redis Streams, Celery, RQ, RabbitMQ

5. Critical inter-service communication

• Requests to other services need retries and delivery guarantees
• Use HTTP/gRPC with retries, or a real message broker

The decision table

The difference from Redis Streams (a preview)

Redis Streams (introduced in Redis 5.0) is the "evolution" of Pub/Sub with persistence:

Streams are out of scope for this guide. If you need the guarantees Streams gives you (delivery guarantees, replay), invest time in learning it.

Troubleshooting

Problem 1: PUBLISH returns 0 (no subscribers)

Cause: There are no subscribers listening on that channel at that moment.

Solution:

1. Check the active subscribers:

   redis-cli> PUBSUB NUMSUB news
   1) "news"
   2) (integer) 0   ← zero subscribers
   

2. If you expect there to be subscribers, debug:

   • Is the subscriber connecting to the same Redis?
   • Is it subscribed to the exact channel (it's case-sensitive)?
   • Did the subscriber service crash?

3. If there may legitimately be no subscribers (that's fine), document it in the code:

   # Pub/Sub: if there are no subscribers, the event is lost.
   # That's OK here because cache invalidation is eventually consistent.
   await r.publish("cache:invalidate", ...)

Problem 2: SUBSCRIBE blocks but receives nothing

Cause: You're subscribed to a different channel from the one they're publishing to.

Solution:

# The publisher
await r.publish("my-channel", "data")

# The subscriber (a typo!)
await pubsub.subscribe("my_channel")  # an underscore, not a dash

Check that the channel names match exactly. Pub/Sub is case-sensitive and does no fuzzy matching.

Problem 3: PSUBSCRIBE gets the confirmation but no messages

Cause: Messages matching a pattern have type="pmessage", not "message".

Solution:

async for msg in pubsub.listen():
    if msg["type"] == "pmessage":   # ← NOT "message" for PSUBSCRIBE
        # process

Problem 4: The subscriber is blocked on a heavy message

Cause: You're processing the message synchronously inside the listen loop.

Solution: Process it in an asynchronous task:

async def process_message(data):
    # Heavy processing
    ...


async for msg in pubsub.listen():
    if msg["type"] == "message":
        # Don't block the listen loop
        asyncio.create_task(process_message(msg["data"]))

This lets the subscriber keep receiving messages while it processes them in parallel.

Problem 5: The connection to Redis closes after a few minutes

Cause: redis-py's default timeout is low, or Redis closes inactive connections.

Solution:

r = Redis(
    host='localhost',
    port=6379,
    decode_responses=True,
    socket_keepalive=True,
    socket_keepalive_options={
        # TCP_KEEPINTVL, TCP_KEEPCNT, TCP_KEEPIDLE
    }
)

For a long-running Pub/Sub, consider a reconnection strategy:

async def subscriber_with_reconnect():
    while True:
        try:
            r = Redis(host='localhost', port=6379, decode_responses=True)
            pubsub = r.pubsub()
            await pubsub.subscribe("channel")

            async for msg in pubsub.listen():
                # process
                pass
        except (ConnectionError, RedisError) as e:
            logger.error(f"Pub/Sub failed: {e}. Reconnecting in 5s...")
            await asyncio.sleep(5)

Problem 6: Messages arrive in an unexpected order

Cause: Pub/Sub does NOT guarantee strict ordering across different channels.

Solution: If you need ordering, use a single channel and sequence your events:

# Fine
await r.publish("events", json.dumps({"type": "A", ...}))
await r.publish("events", json.dumps({"type": "B", ...}))
# The subscriber receives A, B in order (the same channel)

# But this has NO guaranteed ordering:
await r.publish("events:A", "...")
await r.publish("events:B", "...")
# A subscriber with PSUBSCRIBE can receive B before A

If ordering matters, use a single channel.

Exercises

Exercise 1: A basic subscriber + publisher (Easy)

Create two scripts: publisher.py, which publishes 10 messages with a 1s delay, and subscriber.py, which receives them. Run the subscriber in one terminal, the publisher in another, and verify they arrive in order.

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()
    await pubsub.subscribe("test")
    print("Listening on 'test'...")

    async for msg in pubsub.listen():
        if msg["type"] == "message":
            print(f"  Received: {msg['data']}")


asyncio.run(main())

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)

    for i in range(10):
        n = await r.publish("test", f"Message {i}")
        print(f"Published {i} ({n} subscribers)")
        await asyncio.sleep(1)

    await r.aclose()


asyncio.run(main())

# Terminal 1
python subscriber.py

# Terminal 2 (once terminal 1 is listening)
python publisher.py

Exercise 2: A pattern subscription (Easy-Medium)

Modify the subscriber to use PSUBSCRIBE with the pattern events:*. The publisher sends messages to events:login, events:logout, and events:purchase. Verify the subscriber receives all 3.

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()
    await pubsub.psubscribe("events:*")
    print("Pattern-listening on 'events:*'...")

    async for msg in pubsub.listen():
        if msg["type"] == "pmessage":   # ← pmessage, not message
            print(f"  [{msg['pattern']}] {msg['channel']}: {msg['data']}")


asyncio.run(main())

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)

    for event in ["login", "logout", "purchase"]:
        await r.publish(f"events:{event}", f"data-for-{event}")
        print(f"Published events:{event}")

    await r.aclose()


asyncio.run(main())

[events:*] events:login: data-for-login
[events:*] events:logout: data-for-logout
[events:*] events:purchase: data-for-purchase

Exercise 3: Event-driven cache invalidation (Medium)

Implement the real use case: an update_product endpoint that publishes an event, and a subscriber that invalidates the related caches (cache:product:{id} and cache:products:list).

# admin_service.py
import asyncio
import json
import time
from redis.asyncio import Redis


r = Redis(host='localhost', port=6379, decode_responses=True)


async def update_product(product_id: int, data: dict):
    # 1. Update the DB (mock)
    print(f"Updated product:{product_id} with {data}")

    # 2. Publish the invalidation event
    n = await r.publish(
        f"cache:invalidate:product:{product_id}",
        json.dumps({
            "product_id": product_id,
            "fields": list(data.keys()),
            "timestamp": time.time(),
        })
    )
    print(f"Invalidation event published (received by {n} subscribers)")


async def main():
    await asyncio.sleep(2)  # give the subscriber time to connect
    await update_product(42, {"price": 199.99, "stock": 5})
    await asyncio.sleep(0.5)
    await update_product(99, {"price": 49.99})

    await r.aclose()


asyncio.run(main())

# api_service.py (the subscriber)
import asyncio
import json
from redis.asyncio import Redis


r = Redis(host='localhost', port=6379, decode_responses=True)


# Setup: populate the cache with test data
async def setup_cache():
    await r.set("cache:product:42", '{"id":42,"name":"Old"}', ex=300)
    await r.set("cache:product:99", '{"id":99,"name":"Old"}', ex=300)
    await r.set("cache:products:list", '[]', ex=300)
    print("Initial cache populated")


async def handle_invalidation(message):
    """Process an invalidation event."""
    event = json.loads(message["data"])
    product_id = event["product_id"]

    # Invalidate related caches
    await r.delete(f"cache:product:{product_id}")
    await r.delete("cache:products:list")
    print(f"  Invalidated cache:product:{product_id} and cache:products:list")


async def listener():
    pubsub = r.pubsub()
    await pubsub.psubscribe("cache:invalidate:product:*")
    print("API service listening for invalidations...")

    async for msg in pubsub.listen():
        if msg["type"] == "pmessage":
            await handle_invalidation(msg)


async def main():
    await setup_cache()
    await listener()


asyncio.run(main())

# Terminal 1
python api_service.py

# Terminal 2 (wait 1 second)
python admin_service.py

# Terminal 1
Initial cache populated
API service listening for invalidations...
  Invalidated cache:product:42 and cache:products:list
  Invalidated cache:product:99 and cache:products:list

# Terminal 2
Updated product:42 with {'price': 199.99, 'stock': 5}
Invalidation event published (received by 1 subscribers)
...

Exercise 4: A subscriber with reconnect (Medium-Hard)

Implement a subscriber that reconnects automatically if Redis fails. Test: start the subscriber, stop Redis with docker stop redis-dev, wait 10 sec, restart it with docker start redis-dev, and verify the subscriber keeps working.

import asyncio
import logging
from redis.asyncio import Redis
from redis.exceptions import ConnectionError, RedisError


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


async def subscriber_with_reconnect(channel: str, retry_delay: int = 5):
    """A subscriber with automatic reconnection."""
    while True:
        try:
            logger.info(f"Connecting to Redis...")
            r = Redis(host='localhost', port=6379, decode_responses=True, socket_timeout=2)
            await r.ping()  # verify the connection
            logger.info("Connected. Subscribing...")

            pubsub = r.pubsub()
            await pubsub.subscribe(channel)

            async for msg in pubsub.listen():
                if msg["type"] == "message":
                    logger.info(f"  Message: {msg['data']}")

        except (ConnectionError, RedisError) as e:
            logger.warning(f"Redis unavailable: {e}")
            logger.info(f"Retrying in {retry_delay}s...")
            await asyncio.sleep(retry_delay)
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            await asyncio.sleep(retry_delay)
        finally:
            try:
                await r.aclose()
            except Exception:
                pass


async def main():
    await subscriber_with_reconnect("test")


asyncio.run(main())

# Terminal 1
python subscriber_reconnect.py
# Output: Connected. Subscribing...

# Terminal 2
docker stop redis-dev
# Terminal 1: Redis unavailable: Connection refused...
# Terminal 1: Retrying in 5s...

docker start redis-dev
# Terminal 1 (after the retry): Connected. Subscribing...

Exercise 5: A clear limitation — fire and forget (Medium)

Demonstrate Pub/Sub's limitation 1. Publish 5 messages with NO subscribers connected. Then start a subscriber. Verify the subscriber does NOT receive the earlier messages.

import asyncio
from redis.asyncio import Redis


async def step1_publish_with_no_subscribers():
    """Publish 5 messages with no subscribers."""
    r = Redis(host='localhost', port=6379, decode_responses=True)

    for i in range(5):
        n = await r.publish("orphan_channel", f"Historical message #{i}")
        print(f"Published #{i}: {n} subscribers received")
        # n == 0 means the message was LOST

    await r.aclose()


async def step2_subscribe_after():
    """After publishing, subscribe and verify nothing arrives."""
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()
    await pubsub.subscribe("orphan_channel")
    print("\nSubscribed AFTER the publishes. Waiting 3 seconds for messages...")

    try:
        # Wait with a timeout
        await asyncio.wait_for(
            wait_for_message(pubsub),
            timeout=3.0
        )
    except asyncio.TimeoutError:
        print("Timeout. NO message was received (the previous 5 were lost).")
    finally:
        await r.aclose()


async def wait_for_message(pubsub):
    async for msg in pubsub.listen():
        if msg["type"] == "message":
            return msg["data"]


async def main():
    await step1_publish_with_no_subscribers()
    print("\n--- 1 second later ---\n")
    await asyncio.sleep(1)
    await step2_subscribe_after()


asyncio.run(main())

Published #0: 0 subscribers received
Published #1: 0 subscribers received
Published #2: 0 subscribers received
Published #3: 0 subscribers received
Published #4: 0 subscribers received

--- 1 second later ---

Subscribed AFTER the publishes. Waiting 3 seconds for messages...
Timeout. NO message was received (the previous 5 were lost).

Exercise 6: Async task processing inside the listen loop (Hard)

Implement a subscriber that processes "heavy" messages (a 2s sleep simulating work) without blocking the listen loop. Verify you receive new messages while you're processing older ones.

import asyncio
import time
from redis.asyncio import Redis


async def process_message(message_data: str):
    """Simulated heavy work."""
    print(f"  Started processing: {message_data}")
    await asyncio.sleep(2)  # work
    print(f"  Finished: {message_data}")


async def listener():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    pubsub = r.pubsub()
    await pubsub.subscribe("heavy_jobs")

    print("Listening for heavy jobs...")
    async for msg in pubsub.listen():
        if msg["type"] != "message":
            continue

        # Do NOT process it inline (it would block the loop)
        # DO: create an asynchronous task
        asyncio.create_task(process_message(msg["data"]))


async def publisher():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    await asyncio.sleep(1)  # wait for the subscriber

    for i in range(5):
        await r.publish("heavy_jobs", f"job-{i}")
        print(f"Published job-{i}")
        await asyncio.sleep(0.3)  # publish faster than the processing

    await r.aclose()


async def main():
    # Run both concurrently
    await asyncio.gather(
        listener(),
        publisher(),
    )


# Note: listener() runs forever, so in production we'd use another pattern
# For this exercise, run it with a timeout
async def with_timeout():
    try:
        await asyncio.wait_for(main(), timeout=8)
    except asyncio.TimeoutError:
        pass


asyncio.run(with_timeout())

Listening for heavy jobs...
Published job-0
  Started processing: job-0
Published job-1
  Started processing: job-1     ← it starts while job-0 is still processing
Published job-2
  Started processing: job-2
Published job-3
  Started processing: job-3
Published job-4
  Started processing: job-4
  Finished: job-0    ← they finish ~2s after each start
  Finished: job-1
  Finished: job-2
  Finished: job-3
  Finished: job-4

sem = asyncio.Semaphore(3)

async def process_with_limit(data):
    async with sem:
        await process_message(data)

Summary

In this capsule you learned:

The basic commands:

• PUBLISH channel message — publishes a message
• SUBSCRIBE channel — listens to specific channels
• PSUBSCRIBE pattern — listens to channels with wildcards (*, ?, [abc])

The message structure in Python:

• type: "message" (subscribe) or "pmessage" (psubscribe)
• channel: the message's channel
• data: the content
• pattern: the pattern (psubscribe only)

Real use cases:

• Event-driven cache invalidation: decoupled services invalidate caches when data changes
• Real-time notifications: a WebSocket bridge for chat, collaborative editing, live dashboards
• Broadcasting config changes: notifying workers of changes

Critical limitations:

• Fire-and-forget — with no subscribers, the message is lost
• No replay — new subscribers don't receive earlier messages
• No acknowledgment — the publisher doesn't know whether it was processed
• Downed subscribers lose everything
• Distribution to ALL — it isn't a task queue (every subscriber receives everything)

When to use Pub/Sub:

• ✅ Cache invalidation, lightweight notifications, broadcasting config
• ❌ Transactions, payments, audit logs, job queues, event sourcing

Important patterns:

• Process messages with asyncio.create_task() so you don't block the listen loop
• A subscriber with auto-reconnect for resilience
• A clear distinction between message (subscribe) and pmessage (psubscribe)
• PUBSUB NUMSUB channel to check the active subscribers

The 2026 stack: redis.asyncio (included in redis-py >= 4.2), NOT the deprecated aioredis.

Additional resources

1. Redis Pub/Sub Documentation — The official docs
2. redis-py Pub/Sub Examples — Official examples with async
3. Redis Streams vs Pub/Sub — When to use each one
4. Pub/Sub vs Message Queue — A comparison of the paradigms
5. 12 Factor App: Backing Services — Principles for services like Redis/Pub/Sub in production
6. Designing Data-Intensive Applications (Kleppmann), Ch. 11 — The chapter on stream processing — the gold-standard reference

What's next?

In Capsule 03 you go deep into redis.asyncio and connection pooling. You'll learn how to configure the pool correctly, the singleton pattern for sharing the pool across every worker, and why pooling isn't optional in production. It's the technical foundation that module 4's capsule 04 (FastAPI integration) and all of module 5 will assume.

Keep Redis running. Let's go.