Chapter 11 — Pub/Sub & Messaging

Part III — Recipes (Leveraging Redis)

← Chapter 10 — Caching Patterns | Table of Contents | Chapter 12 — Search & Analytics →

Redis Pub/Sub is a fire-and-forget broadcast channel: publishers send messages to a channel name; subscribers receive them in real time. It is ideal for live notifications, chat fan-out, and cache invalidation signals. It is not a durable message queue—messages are dropped if no subscriber is listening.

This chapter builds a chat room on Pub/Sub, explores pattern subscriptions, explains when Pub/Sub falls short, and shows how Streams and SSE/WebSocket bridges deliver production-grade messaging.

When to reach for each primitive:

Need	Use
Live broadcast, loss OK	Pub/Sub
Pattern-based fan-out	PSUBSCRIBE
Durable log, replay	Streams
Competing workers	Streams + consumer groups
Browser clients	Gateway + SSE/WebSocket

Recipe 11.1 — Chat Room with PUBLISH/SUBSCRIBE

Problem

You need a simple multi-user chat where messages broadcast instantly to everyone in a room. Latency must stay under tens of milliseconds; persistence is optional for now.

Solution

Each chat room maps to a Redis channel. Clients subscribe to chat:room:{id}; sending a message is a PUBLISH.

redis-cli — two terminal sessions:

Terminal A (subscriber):

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

Terminal B (publisher):

127.0.0.1:6379> PUBLISH chat:room:lobby '{"user":"alice","text":"Hello!"}'
(integer) 1

Terminal A receives:

1) "message"
2) "chat:room:lobby"
3) "{\"user\":\"alice\",\"text\":\"Hello!\"}"

The integer return value of PUBLISH is the count of clients subscribed to that channel (across all connections on this Redis node).

Python — subscriber in a background thread:

import json
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
pubsub = r.pubsub()
pubsub.subscribe("chat:room:lobby")

for message in pubsub.listen():
    if message["type"] == "message":
        payload = json.loads(message["data"])
        print(f"{payload['user']}: {payload['text']}")

Python — publish:

def send_message(room: str, user: str, text: str):
    r.publish(f"chat:room:{room}", json.dumps({"user": user, "text": text}))

Node.js — subscriber:

const Redis = require('ioredis');
const sub = new Redis();
const pub = new Redis();

sub.subscribe('chat:room:lobby', (err, count) => {
  console.log(`Subscribed to ${count} channel(s)`);
});

sub.on('message', (channel, message) => {
  const { user, text } = JSON.parse(message);
  console.log(`${user}: ${text}`);
});

async function sendMessage(room, user, text) {
  await pub.publish(`chat:room:${room}`, JSON.stringify({ user, text }));
}

Go — publish:

func SendMessage(ctx context.Context, rdb *redis.Client, room, user, text string) error {
    payload, _ := json.Marshal(map[string]string{"user": user, "text": text})
    return rdb.Publish(ctx, "chat:room:"+room, payload).Err()
}

Go subscribers use PubSub.ReceiveMessage in a loop; use a dedicated connection—Pub/Sub connections cannot interleave other commands.

Discussion

SUBSCRIBE puts the connection into subscriber mode until UNSUBSCRIBE or disconnect.
Use a separate connection for publishing while subscribed (all clients follow this rule).
Messages have no IDs, no ACK, no replay. Offline users miss messages.
Payloads should stay small (under 1 KB ideally); Pub/Sub shares the single-threaded event loop with other commands.

For chat history, persist messages separately (PostgreSQL, or Streams—Recipe 11.4).

Multi-room presence (optional extension):

Track who is online with a Set per room and a TTL heartbeat:

127.0.0.1:6379> SADD chat:room:lobby:online user:alice
127.0.0.1:6379> SET chat:presence:alice 1 EX 60
127.0.0.1:6379> SMEMBERS chat:room:lobby:online

A background sweeper removes users whose presence key expired. Publish join/leave events on chat:room:lobby:events for UI updates.

Recipe 11.2 — Pattern Subscribe (PSUBSCRIBE)

Problem

A moderation dashboard must listen to all chat rooms (chat:room:*) without subscribing to each channel individually. A microservice needs every cache-invalidation event matching cache:invalidate:product:*.

Solution

PSUBSCRIBE uses glob-style patterns:

127.0.0.1:6379> PSUBSCRIBE chat:room:*
Reading messages...
1) "psubscribe"
2) "chat:room:*"
3) (integer) 1

When someone publishes to chat:room:engineering:

1) "pmessage"
2) "chat:room:*"
3) "chat:room:engineering"
4) "{\"user\":\"bob\",\"text\":\"deploy done\"}"

Note the four-element response: event type, pattern, actual channel, payload.

Python:

pubsub = r.pubsub()
pubsub.psubscribe("chat:room:*")

for message in pubsub.listen():
    if message["type"] == "pmessage":
        channel = message["channel"]       # e.g. chat:room:engineering
        room = channel.split(":")[-1]
        payload = json.loads(message["data"])
        print(f"[{room}] {payload['user']}: {payload['text']}")

Node.js:

sub.psubscribe('cache:invalidate:product:*');
sub.on('pmessage', (pattern, channel, message) => {
  console.log(`Pattern ${pattern} matched ${channel}: ${message}`);
});

Go:

pubsub := rdb.PSubscribe(ctx, "chat:room:*")
for {
    msg, err := pubsub.ReceiveMessage(ctx)
    if err != nil {
        break
    }
    // msg.Channel, msg.Payload
}

Discussion

Patterns add CPU cost on the server—Redis matches every publish against active patterns. Keep pattern count modest (dozens, not thousands).

Prefer exact channels when you know the set of rooms; use patterns for open-ended namespaces.

PUNSUBSCRIBE without arguments drops all pattern subscriptions on that connection.

Recipe 11.3 — Pub/Sub Limitations (No Persistence)

Problem

You chose Pub/Sub for order notifications. A consumer restarts for deploy—and misses every message published during the 30-second downtime. Product asks: "Why did we lose events?"

Solution

Understand what Pub/Sub does not guarantee:

Property	Pub/Sub	Durable queue (Streams, Kafka)
Persistence	None — fire-and-forget	Messages stored until trimmed/ACKed
Offline delivery	Missed forever	Consumer catches up from offset
At-least-once	No	Yes (with ACK/retry)
Fan-out	Native, cheap	Consumer groups per subscriber set
Ordering	Per channel, best-effort	Per stream partition, guaranteed

Demonstration — publish with zero subscribers:

127.0.0.1:6379> PUBLISH events:orders '{"id":9001}'
(integer) 0

Return value 0 means no one received it. The message is gone.

When Pub/Sub is still correct:

Live UI updates (presence, typing indicators).
Cache invalidation (miss one → stale until next write).
Triggering idempotent refresh ("something changed" signal).
Low-latency broadcast where loss is acceptable.

When to leave Pub/Sub:

Payment webhooks, audit trails, job dispatch.
Any "every message must be processed exactly once" requirement.

Migrate signal paths to Redis Streams (Recipe 11.4) or an external broker; keep Pub/Sub only for the live UI fan-out layer.

Discussion

Pub/Sub shares Redis's memory but not its persistence story—Pub/Sub messages never touch the RDB/AOF. Replication does not replay Pub/Sub to replicas' subscribers in a way that helps offline consumers.

Cluster note: Pub/Sub in Cluster mode routes channel messages across the cluster (Redis 7+ improved sharded pub/sub with SPUBLISH/SSUBSCRIBE for slot-local channels). Design channel names with cluster constraints in mind.

Recipe 11.4 — Streams for Durable Messaging

Problem

You need reliable order-event delivery: consumers can crash, restart, and continue from where they left off. Multiple worker instances must share load without duplicate processing (with ACK semantics).

Solution

Use a Stream as an append-only log. Producers XADD; consumers join a consumer group and read with XREADGROUP.

redis-cli:

127.0.0.1:6379> XADD events:orders * order_id 9001 status "created" amount 49.99
"1716892800000-0"

127.0.0.1:6379> XGROUP CREATE events:orders fulfillment $ MKSTREAM
OK

127.0.0.1:6379> XREADGROUP GROUP fulfillment worker-1 COUNT 10 BLOCK 5000 STREAMS events:orders >
1) 1) "events:orders"
   2) 1) 1) "1716892800000-0"
         2) 1) "order_id"
            2) "9001"
            3) "status"
            4) "created"
            5) "amount"
            6) "49.99"

127.0.0.1:6379> XACK events:orders fulfillment 1716892800000-0
(integer) 1

Python — producer and consumer:

def publish_order_event(r, order_id, status, amount):
    return r.xadd("events:orders", {
        "order_id": order_id,
        "status": status,
        "amount": str(amount),
    })

def consume_orders(r, group="fulfillment", consumer="worker-1"):
    try:
        r.xgroup_create("events:orders", group, id="$", mkstream=True)
    except redis.ResponseError as e:
        if "BUSYGROUP" not in str(e):
            raise

    while True:
        messages = r.xreadgroup(
            group, consumer,
            {"events:orders": ">"},
            count=10, block=5000,
        )
        for stream, entries in messages:
            for msg_id, fields in entries:
                process_order(fields)
                r.xack(stream, group, msg_id)

Node.js:

await redis.xadd('events:orders', '*', 'order_id', '9001', 'status', 'created');

const results = await redis.xreadgroup(
  'GROUP', 'fulfillment', 'worker-1',
  'COUNT', '10', 'BLOCK', '5000',
  'STREAMS', 'events:orders', '>'
);

Go:

id, err := rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "events:orders",
    Values: map[string]interface{}{"order_id": "9001", "status": "created"},
}).Result()

streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "fulfillment",
    Consumer: "worker-1",
    Streams:  []string{"events:orders", ">"},
    Count:    10,
    Block:    5 * time.Second,
}).Result()

Handle pending entries with XPENDING and XCLAIM for stale consumers—see Chapter 13.

Discussion

Streams give you persistence (AOF/RDB), consumer groups, and replay. Trade-offs vs Pub/Sub:

Higher memory (messages stored until XTRIM or MAXLEN).
Slightly higher latency than Pub/Sub.
Consumer groups add operational complexity (PEL management).

Hybrid architecture: XADD for durability, then PUBLISH a lightweight "new message" ping so live UIs wake up instantly without polling.

Dual-write pattern (Python):

def publish_order_event_hybrid(r, order_id, status, amount):
    msg_id = r.xadd("events:orders", {
        "order_id": order_id, "status": status, "amount": str(amount),
    })
    r.publish("events:orders:notify", json.dumps({"id": msg_id, "order_id": order_id}))
    return msg_id

Live dashboards subscribe to events:orders:notify; reconciliation workers consume the Stream.

Recipe 11.5 — SSE/WebSocket Bridge Pattern

Problem

Browsers cannot open a raw Redis Pub/Sub connection. You need a web gateway that forwards Redis events to clients over Server-Sent Events (SSE) or WebSockets.

Solution

Architecture:

Browser ←SSE/WS→ Gateway (Node/Go) ←Pub/Sub or Streams→ Redis

The gateway subscribes to Redis; browser clients connect to HTTP/WebSocket endpoints on the gateway.

Node.js — SSE bridge with ioredis:

const express = require('express');
const Redis = require('ioredis');

const app = express();
const sub = new Redis();

const clients = new Map(); // room -> Set<res>

sub.psubscribe('chat:room:*');
sub.on('pmessage', (pattern, channel, message) => {
  const room = channel.replace('chat:room:', '');
  const listeners = clients.get(room);
  if (!listeners) return;
  for (const res of listeners) {
    res.write(`data: ${message}\n\n`);
  }
});

app.get('/chat/:room/stream', (req, res) => {
  const { room } = req.params;
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.flushHeaders();

  if (!clients.has(room)) clients.set(room, new Set());
  clients.get(room).add(res);

  req.on('close', () => {
    clients.get(room)?.delete(res);
  });
});

app.listen(3000);

Python — FastAPI SSE sketch:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()
message_queue: asyncio.Queue = asyncio.Queue()

def redis_listener():
    pubsub = r.pubsub()
    pubsub.psubscribe("chat:room:*")
    for msg in pubsub.listen():
        if msg["type"] == "pmessage":
            asyncio.get_event_loop().call_soon_threadsafe(
                message_queue.put_nowait, msg
            )

@app.get("/chat/{room}/stream")
async def stream(room: str):
    async def event_generator():
        while True:
            msg = await message_queue.get()
            if msg["channel"] == f"chat:room:{room}":
                yield f"data: {msg['data']}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Go — WebSocket hub pattern:

Use gorilla/websocket or nhooyr.io/websocket. One goroutine runs PubSub.ReceiveMessage; fan-out to connected clients via channels. Protect against slow clients with bounded buffers or drop-on-backpressure.

Publishing from the web tier:

// POST /chat/:room — HTTP handler
await pub.publish(`chat:room:${room}`, JSON.stringify({ user, text }));

Discussion

SSE is simpler (one-way, HTTP/2 friendly, auto-reconnect). WebSockets suit bidirectional games or collaborative editing.
Scale gateways horizontally: each instance subscribes to the same Redis channels—clients may land on any gateway via load balancer sticky sessions or shared room routing.
For guaranteed delivery to browsers, persist in Streams; on reconnect, client fetches history via REST then attaches SSE for live updates.
Authenticate room subscriptions—Redis Pub/Sub has no ACL per channel at the message level; enforce auth in the gateway.

Chapter Summary

Recipe	Mechanism	Durability	Use case
11.1	SUBSCRIBE/PUBLISH	None	Live chat, notifications
11.2	PSUBSCRIBE	None	Multi-room dashboards
11.3	Limitations	—	Know when not to use Pub/Sub
11.4	Streams + groups	Yes	Order events, audit, workers
11.5	SSE/WS bridge	Gateway-dependent	Browser clients

Pub/Sub is Redis's megaphone. Streams is its ledger. Most production systems use both: Streams for truth, Pub/Sub for speed.

Operational Notes for Messaging at Scale

Connection budgeting

Every subscriber holds an open TCP connection. A chat service with 50,000 concurrent WebSocket clients might run 20 gateway instances each holding one Redis Pub/Sub connection—manageable. If each browser connected directly to Redis (never do this), you'd exhaust file descriptors instantly. Plan gateway count × Redis connections, plus publisher pools.

Message size and serialization

Prefer JSON for debugging; use MessagePack or Protocol Buffers if bandwidth matters. Enforce max payload size in your gateway before PUBLISH. Large messages block the Redis main thread longer than small ones.

Monitoring

Track:

PUBLISH rate per channel (application metrics).
Subscriber client count via CLIENT LIST filtered by cmd=subscribe.
Gateway fan-out latency (Redis receive → browser SSE flush).

Alert when publish rate exceeds subscriber consumption—backpressure in the gateway, not Redis, is usually the bottleneck.

Sharded Pub/Sub in Cluster (Redis 7.0+)

In Redis Cluster, channel names hash to slots. SPUBLISH / SSUBSCRIBE restrict pub/sub traffic to a slot's master, reducing cross-node fan-out for slot-local events:

127.0.0.1:6379> SPUBLISH notifications:shard{user1000} "payment received"

Use hash tags in channel names when colocating related data and notifications on one slot.

Testing Pub/Sub locally

Use two redis-cli sessions for quick verification, or a minimal script:

# test_pubsub.py — verify round-trip under 10ms locally
import threading, time, json, redis

received = []
def sub():
    ps = r.pubsub()
    ps.subscribe("test:ping")
    for m in ps.listen():
        if m["type"] == "message":
            received.append(time.time())

r = redis.Redis(decode_responses=True)
threading.Thread(target=sub, daemon=True).start()
time.sleep(0.1)
t0 = time.time()
r.publish("test:ping", json.dumps({"t": t0}))
time.sleep(0.1)
assert received, "No message received"
print(f"Latency: {(received[0] - t0) * 1000:.1f} ms")

← Chapter 10 — Caching Patterns | Table of Contents | Chapter 12 — Search & Analytics →

Navigation Menu

Chapter 11 — Pub/Sub & Messaging

Recipe 11.1 — Chat Room with PUBLISH/SUBSCRIBE

Problem

Solution

Discussion

See Also

Recipe 11.2 — Pattern Subscribe (PSUBSCRIBE)

Problem

Solution

Discussion

See Also

Recipe 11.3 — Pub/Sub Limitations (No Persistence)

Problem

Solution

Discussion

See Also

Recipe 11.4 — Streams for Durable Messaging

Problem

Solution

Discussion

See Also

Recipe 11.5 — SSE/WebSocket Bridge Pattern

Problem

Solution

Discussion

See Also

Chapter Summary

Operational Notes for Messaging at Scale

Connection budgeting

Message size and serialization

Monitoring

Sharded Pub/Sub in Cluster (Redis 7.0+)

Testing Pub/Sub locally

FilesExpand file tree

11-pubsub-and-messaging.md

Latest commit

History

11-pubsub-and-messaging.md

File metadata and controls

Chapter 11 — Pub/Sub & Messaging

Recipe 11.1 — Chat Room with PUBLISH/SUBSCRIBE

Problem

Solution

Discussion

See Also

Recipe 11.2 — Pattern Subscribe (PSUBSCRIBE)

Problem

Solution

Discussion

See Also

Recipe 11.3 — Pub/Sub Limitations (No Persistence)

Problem

Solution

Discussion

See Also

Recipe 11.4 — Streams for Durable Messaging

Problem

Solution

Discussion

See Also

Recipe 11.5 — SSE/WebSocket Bridge Pattern

Problem

Solution

Discussion

See Also

Chapter Summary

Operational Notes for Messaging at Scale

Connection budgeting

Message size and serialization

Monitoring

Sharded Pub/Sub in Cluster (Redis 7.0+)

Testing Pub/Sub locally