Bài 73: Streams vs Pub/Sub — Khi Nào Dùng Cái Nào

1. Mục Tiêu Bài Học

• Phân biệt rõ semantics at-most-once (Pub/Sub) và at-least-once (Streams) và hậu quả thực tế khi chọn sai.
• Biết trade-off throughput, latency, memory của từng cơ chế.
• Áp dụng được decision tree để chọn Pub/Sub, Streams, hoặc kết hợp cả hai.
• Triển khai hybrid pattern: Stream ghi lịch sử, Pub/Sub giao ngay.
• Nhận ra anti-pattern phổ biến khi trộn lẫn không đúng cách.

2. TL;DR — Quick Reference

3. Semantics Cốt Lõi

Pub/Sub — at-most-once

Pub/Sub trong Redis hoạt động theo kiểu fire-and-forget: message được deliver cho subscriber đang kết nối tại thời điểm publish. Redis không giữ lại bản sao nào sau khi deliver (hoặc khi không có subscriber nào). Hậu quả trực tiếp:

• Subscriber offline → MISS message đó vĩnh viễn. Reconnect sau 1 giây cũng không đọc lại được.
• Redis crash tại thời điểm publish → message mất, không có cơ chế retry.
• Subscriber chậm (slow consumer) → Redis drop message hoặc disconnect subscriber (tùy cấu hình client-output-buffer-limit).

Không có acknowledgement. Publisher không biết message đã được xử lý hay chưa.

Streams — at-least-once

Redis Streams (XADD) lưu message vào cấu trúc dữ liệu trong RAM (và persist ra RDB/AOF nếu bật). Message tồn tại cho đến khi bị xoá bằng XTRIM/XDEL hoặc MAXLEN tự trim. Hậu quả:

• Consumer offline → reconnect và đọc lại từ ID đã lưu (last_delivered_id của consumer group, hoặc cursor tự quản nếu dùng XREAD).
• Consumer crash giữa chừng → message nằm trong PEL (Pending Entry List) của consumer group, chờ XCLAIM hoặc XAUTOCLAIM để reassign.
• XACK báo message đã xử lý xong → xoá khỏi PEL.

At-least-once có nghĩa message có thể được deliver hơn một lần nếu consumer crash sau khi xử lý nhưng trước XACK. Consumer cần idempotent hoặc deduplication (bài 58).

Streams + consumer group — parallel processing

Consumer group chia message cho nhiều consumer xử lý song song — mỗi message chỉ đến một consumer trong group (load balancing). Đây là điểm Pub/Sub không có: Pub/Sub fan-out tới tất cả subscriber, không phân chia.

4. Khi Nào Dùng Pub/Sub

Pub/Sub phù hợp khi message loss có thể chấp nhận được hoặc khi cần fan-out broadcast tới tất cả subscriber.

Use case rõ ràng

• Broadcast notification realtime tới online users: typing indicator (bài 71), presence event (bài 69). Nếu user offline không cần nhận ngay, Pub/Sub đủ dùng; dữ liệu stale sẽ tự recover khi user reconnect query lại DB.
• Cache invalidation event: mất event = cache stale thêm vài giây cho đến khi TTL hết — chấp nhận được với most read-heavy systems. Chi phí persist Stream không xứng đáng.
• Live dashboard updates: client cần giá trị mới nhất, không cần lịch sử. Nếu miss một update, update kế tiếp sẽ đến.
• Inter-instance WebSocket backplane: một app server instance nhận WebSocket message từ client, publish qua Pub/Sub để tất cả instance khác forward tới user tương ứng (bài 67). Message bị miss chỉ xảy ra nếu instance down — hiếm và thường được xử lý ở tầng reconnect.
• Feature flag / config reload broadcast: cần toàn bộ app server nhận, scope global, loss nhỏ chấp nhận được.

Throughput cao hơn vì không persist

PUBLISH không ghi vào bất kỳ cấu trúc dữ liệu nào — chỉ iterate danh sách subscriber và gửi. Không có memory allocation per-message sau khi deliver. Đây là lý do throughput Pub/Sub cao hơn Streams khoảng 20–50% trên cùng hardware.

5. Khi Nào Dùng Streams

Streams phù hợp khi mất message có hậu quả nghiêm trọng, cần audit log, hoặc cần multiple consumer xử lý song song.

Use case rõ ràng

• Job queue async: worker process job, cần ack + retry khi worker crash. PEL tracking đảm bảo không job nào bị mất (bài 53–56).
• Order / payment events: mất event = mất tiền. Consumer cần ack sau khi xử lý thành công. Dùng idempotency key để tránh double-process khi retry.
• Event sourcing / audit log: cần replay toàn bộ event để rebuild state, hoặc để debug. Stream giữ history.
• Chat history: user reconnect sau khi offline cần đọc lại tin nhắn đã bỏ lỡ (bài 72). XREAD từ last_read_id.
• Multi-consumer parallel: xử lý image processing, email sending với N worker. Consumer group phân chia message, không duplicate (bài 54).
• Telemetry / metrics pipeline: batched ingestion, có thể sample với MAXLEN, cần traceability.

Lưu ý MAXLEN bắt buộc

Stream không tự trim. Không đặt MAXLEN thì Stream phình vô hạn trong RAM. Luôn xác định retention policy:

# MAXLEN ~ = approximate trim (nhanh hơn exact trim)
XADD chat:room:abc MAXLEN ~ 10000 * field1 value1

# Hoặc dùng XTRIM định kỳ
XTRIM chat:room:abc MAXLEN ~ 10000

6. Throughput & Latency Benchmark

Số liệu dưới đây là estimate trên single Redis instance (Redis 7.x, hardware thông thường — 4–8 core, NVMe, 10GbE). Kết quả thực tế phụ thuộc payload size, network, và config.

Throughput

Latency p99

Nếu latency <1ms là yêu cầu cứng và message loss được phép, Pub/Sub là lựa chọn đúng. Nếu cần durability, Streams với tail read (1–5ms) thường chấp nhận được với most use case.

7. Memory Footprint

Pub/Sub

Gần như zero per-message. Redis chỉ duy trì danh sách subscriber per channel (hash map). Message không được lưu sau khi deliver. Memory chính của Pub/Sub là connection overhead: mỗi subscriber connection chiếm ~50KB–2MB tùy output buffer.

Streams

Mỗi message chiếm khoảng 100–500 byte tùy số field và payload size. Ngoài ra còn có PEL entry (~100 byte) cho mỗi message đang pending trong consumer group.

Ước tính thực tế:

• 1 triệu message × 200 byte/message ≈ 200MB RAM.
• 1 triệu message × 500 byte/message ≈ 500MB RAM.
• MAXLEN 10.000 messages: tối đa ~1–5MB per stream — manageable.

Kết luận: Stream cho chat room với MAXLEN 10.000 tin nhắn dùng vài MB per room. 10.000 room → vài chục GB. MAXLEN bắt buộc cho production.

8. Decision Tree

Cần subscriber offline nhận lại được không?
├─ KHÔNG → Pub/Sub
│            ↓
│     Cần fan-out broadcast tới TẤT CẢ subscriber?
│     ├─ CÓ  → PUBLISH (regular hoặc sharded tùy Cluster)
│     └─ N/A → PUBLISH (fan-out là default của Pub/Sub)
│
└─ CÓ → Stream
           ↓
    Cần nhiều consumer xử lý song song (load balancing)?
    ├─ CÓ  → XREADGROUP + consumer group + XACK
    └─ KHÔNG → XREAD tail (XREAD BLOCK 0 STREAMS key $)
                  ↓
           Cần replica đọc cùng stream?
           ├─ CÓ  → XREAD với cursor per reader
           └─ KHÔNG → XREAD BLOCK simple

Câu hỏi bổ sung để xác nhận

• Hậu quả của mất một message là gì? Nghiêm trọng → Stream. Chấp nhận được → Pub/Sub.
• Message có cần xử lý bởi nhiều consumer song song không? Có → Stream consumer group.
• Cần lịch sử / replay không? Có → Stream.
• Latency yêu cầu dưới 1ms không? Có → Pub/Sub (hoặc Streams chỉ phù hợp nếu 1–5ms ok).
• Throughput > 500k msg/s không? Có → cân nhắc Pub/Sub hoặc horizontal scale Streams.

9. Pattern Hybrid — Online Realtime + Offline Durable

Nhiều hệ thống cần cả hai: user online nhận message ngay lập tức (low latency), user offline phải nhận lại khi reconnect (durability). Giải pháp: ghi đồng thời vào Stream và Pub/Sub.

Cơ chế

• User online: WebSocket server subscribe channel Pub/Sub → nhận message <1ms.
• User offline: khi reconnect, đọc Stream từ last_read_id → nhận message đã bỏ lỡ.
• Stream là source of truth lịch sử; Pub/Sub là transport realtime.

Code Python redis-py

import json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def send_chat_message(room_id: str, user_id: str, text: str) -> str:
    """
    Ghi message vào cả Stream và Pub/Sub.
    Trả về Stream message ID để client lưu làm cursor.
    """
    payload = {"user_id": user_id, "text": text}

    # 1. Ghi vào Stream — durability + history
    msg_id = r.xadd(
        f"chat:stream:{room_id}",
        payload,
        maxlen=10000,   # ~ trim, approximate
    )

    # 2. Broadcast qua Pub/Sub — realtime delivery
    payload["stream_id"] = msg_id  # kèm ID để client biết cursor
    r.publish(f"chat:channel:{room_id}", json.dumps(payload))

    return msg_id


def load_missed_messages(room_id: str, last_read_id: str) -> list:
    """
    Đọc message từ Stream từ sau last_read_id.
    Gọi khi user reconnect sau khi offline.
    """
    if not last_read_id:
        # Lần đầu hoặc không có cursor — đọc N tin nhắn gần nhất
        entries = r.xrevrange(f"chat:stream:{room_id}", count=50)
        return list(reversed(entries))

    # Đọc từ sau cursor (exclusive)
    entries = r.xrange(f"chat:stream:{room_id}", min=f"({last_read_id}")
    return entries

Trade-off của pattern này

• Write amplification: mỗi message ghi 2 lần (XADD + PUBLISH). I/O tăng gấp đôi — chấp nhận được với most use case vì latency Pub/Sub vẫn thấp.
• Consistency: có window nhỏ giữa XADD thành công và PUBLISH chưa xong. User online có thể nhận Pub/Sub trước khi Stream ghi xong (trong thực tế XADD và PUBLISH cùng thread → sequential, không race). Nếu dùng pipeline, ghi XADD trước.
• Two sources of truth: Pub/Sub message và Stream phải sync. Bao giờ cũng coi Stream là authoritative; Pub/Sub chỉ là delivery channel.

10. Streams Only + Tail Read

Thay vì hybrid, có thể dùng Stream thuần với XREAD BLOCK để đạt gần Pub/Sub semantics — nhưng với durability.

Cách hoạt động

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def stream_tail_subscribe(stream_key: str):
    """
    Đọc message mới từ Stream theo kiểu blocking poll.
    '$' = chỉ đọc message mới từ lúc gọi lệnh này.
    """
    last_id = "$"

    while True:
        # BLOCK 0 = block vô hạn cho đến khi có message mới
        results = r.xread({stream_key: last_id}, block=0)
        if results:
            for _, messages in results:
                for msg_id, fields in messages:
                    yield msg_id, fields
                    last_id = msg_id  # cập nhật cursor

So sánh với Pub/Sub

Với XREAD BLOCK, mỗi reader cần giữ cursor riêng (last_id). Fan-out vẫn hoạt động vì nhiều reader có thể đọc cùng stream song song — khác consumer group (mỗi message chỉ đến một consumer). Overhead thêm ~5× so với Pub/Sub ở throughput cao.

11. Cluster Considerations

Pub/Sub trong Cluster

• Regular Pub/Sub: PUBLISH broadcast qua cluster bus tới tất cả node. Cluster 10 node → mỗi message đi qua 10 lần. Scale kém — bài 68 đã phân tích.
• Sharded Pub/Sub (Redis 7.0+): SPUBLISH route theo hash slot, chỉ ở 1 node. Throughput scale tuyến tính theo số node. Phù hợp cho per-entity channel (per-room, per-user).

Streams trong Cluster

Stream là một key thông thường — thuộc về 1 hash slot → 1 node. Không có cluster bus overhead. Consumer group trên cùng stream hoạt động bình thường. Nếu muốn sharding Stream theo room, cần hash tag:

# Tất cả stream của room_123 về cùng slot
XADD "chat:{room_123}:messages" MAXLEN ~ 10000 * ...
XADD "chat:{room_123}:events"   MAXLEN ~ 1000  * ...

Consumer group với nhiều worker trên Cluster

Worker đọc stream qua XREADGROUP — tất cả worker kết nối đến node owner của stream key. Nếu worker chạy trên nhiều server, các server đều kết nối đến cùng 1 Redis node (node owner của slot đó). Nếu stream là hotspot, cân nhắc sharding ở application layer (nhiều stream riêng).

12. So Sánh Kafka / RabbitMQ

Redis không phải replacement cho Kafka/RabbitMQ trong mọi trường hợp. Bài 60 đã so sánh chi tiết; phần này chỉ tóm tắt positioning.

Redis sweet spot: latency thấp, data đã nằm trong Redis (cache + stream cùng một hệ thống), retention ngắn. Khi cần retention nhiều tuần, multi-datacenter replication, hoặc throughput hàng chục TB/ngày — Kafka phù hợp hơn.

13. Migration Patterns

Pub/Sub → Streams (khi cần persistence)

Kịch bản: hệ thống đang dùng Pub/Sub, phát hiện message loss khi subscriber crash, cần thêm durability.

1. Bắt đầu publish song song: publisher ghi cả PUBLISH và XADD. Consumer vẫn đọc từ Pub/Sub.
2. Consumer switch sang Stream: từng consumer một switch sang XREAD / XREADGROUP, đọc từ Stream. Giai đoạn này có thể có duplicate (nhận từ cả Pub/Sub và Stream) — xử lý bằng idempotency.
3. Tắt PUBLISH: sau khi tất cả consumer đã chuyển sang Stream, dừng PUBLISH. Xác nhận không còn subscriber qua PUBSUB CHANNELS và INFO clients (pub_sub_channels).

Streams → Pub/Sub (khi throughput không đủ)

Hiếm gặp hơn vì thường khi cần throughput cao hơn, câu trả lời là scale Redis hoặc sharding — không phải mất durability. Nếu thực sự cần:

1. Verify semantics: xác nhận message loss là chấp nhận được với business owner.
2. Chạy song song Pub/Sub subscriber trước, đo latency và throughput.
3. Tắt XADD khi Pub/Sub đã ổn định.
4. Cleanup: XTRIM hoặc DEL Stream key khi không còn reader.

14. Anti-patterns

Pub/Sub cho critical events

Dùng Pub/Sub cho payment event, order event, hoặc bất kỳ event nào mà loss = mất dữ liệu nghiệp vụ. Subscriber crash → event mất → giao dịch không được xử lý. Fix: Stream + consumer group + XACK.

Stream cho ephemeral signal mà không trim

Dùng Stream cho cache invalidation signal (chỉ cần trigger, không cần history) mà không đặt MAXLEN. Signal này rẻ bằng Pub/Sub nhưng tốn RAM không cần thiết. Nếu đã dùng Stream, bắt buộc MAXLEN phù hợp (ví dụ MAXLEN ~ 1000 cho cache invalidation).

PEL grow vô hạn

Dùng XREADGROUP nhưng không gọi XACK. Mỗi message được đọc sẽ nằm trong PEL mãi mãi. PEL grow → XPENDING trả về list khổng lồ → XAUTOCLAIM chậm. Fix: XACK ngay sau khi xử lý; nếu consumer crash, dùng XAUTOCLAIM (bài 56).

Pub/Sub subscriber chậm (slow consumer)

Publisher gửi nhanh hơn subscriber xử lý → output buffer của subscriber connection phình to. Redis có thể disconnect subscriber khi buffer vượt client-output-buffer-limit pubsub. Lúc này subscriber MISS message và cũng bị ngắt kết nối. Fix: tăng buffer limit hoặc tối ưu subscriber; hoặc switch sang Stream nếu throughput subscriber thực sự không kịp.

# redis.conf — default pubsub output buffer limit
# client-output-buffer-limit pubsub 32mb 8mb 60
# 32mb hard limit, hoặc 8mb sustained > 60 giây → disconnect
# Tăng nếu subscriber chậm và loss không chấp nhận được

Hai nguồn truth (Pub/Sub + Stream) không đồng bộ

Pattern hybrid ghi XADD và PUBLISH theo thứ tự bị đảo: PUBLISH trước, XADD thất bại. User online nhận Pub/Sub nhưng history trong Stream thiếu message → inconsistency. Fix: luôn ghi Stream trước, Pub/Sub sau. Nếu PUBLISH thất bại, message vẫn an toàn trong Stream (user sẽ đọc qua load_missed_messages khi cần).

Stream với 10M consumer group member

Consumer group không phù hợp với hàng triệu consumer vì overhead PEL per consumer, XAUTOCLAIM cần scan PEL, XPENDING trả về kết quả lớn. Fan-out tới hàng triệu người dùng vẫn nên dùng Pub/Sub hoặc push notification + Stream cho history.

15. Checklist Quyết Định

Trả lời từng câu, kết quả quyết định lựa chọn:

• Cần persist message lâu hơn 5 phút? → Stream.
• Subscriber có thể offline và cần nhận lại khi reconnect? → Stream.
• Cần ack + retry (không muốn mất message khi consumer crash)? → Stream + XACK.
• Cần nhiều consumer xử lý song song (load balancing)? → Stream + consumer group.
• Cần broadcast tới tất cả subscriber đồng thời? → Pub/Sub (fan-out native).
• Subscriber join/leave động, số lượng không biết trước? → Pub/Sub (SUBSCRIBE/UNSUBSCRIBE đơn giản hơn quản lý consumer group).
• Broadcast tới hàng triệu subscriber trên Redis Cluster? → Sharded Pub/Sub (Redis 7.0+).
• Latency yêu cầu dưới 1ms p99? → Pub/Sub (hoặc chấp nhận 1–5ms với XREAD BLOCK).
• Cần exactly-once semantics? → Stream + idempotency key (at-least-once + dedup).
• Message loss = hậu quả nghiêm trọng (payment, order, audit)? → Stream, không dùng Pub/Sub.

Pattern hybrid phổ biến

• Cache invalidation broadcast: Pub/Sub (loss = stale vài giây, chấp nhận).
• User notification: Stream (durable) + push notification (FCM/APNs) cho offline. Pub/Sub cho online delivery.
• WebSocket message: Pub/Sub realtime backplane + DB/Stream persist.
• Chat: Stream history + Pub/Sub delivery (hybrid bài 9).
• Telemetry: Stream với MAXLEN sample, batched consumer.

16. Operational Concerns

Monitor Pub/Sub

# Số channel đang active và subscriber count
INFO clients
# → pub_sub_channels: N
# → pub_sub_patterns: N

# Liệt kê channel đang có subscriber
PUBSUB CHANNELS *

# Đếm subscriber của channel cụ thể
PUBSUB NUMSUB chat:room:abc

# Detect slow subscriber (output buffer lớn)
CLIENT LIST
# → kết quả có: omem=XXXXX (output buffer size bytes)

Monitor Streams

# Thông tin stream (length, first/last ID, groups)
XINFO STREAM chat:stream:abc

# Thông tin consumer group (pending, last-delivered-id)
XINFO GROUPS chat:stream:abc

# PEL của consumer group
XPENDING chat:stream:abc my-group - + 10

# Consumer cụ thể trong group
XINFO CONSUMERS chat:stream:abc my-group

Alert cần thiết cho production

• Pub/Sub: pub_sub_channels tăng đột biến (có thể là memory leak subscriber); omem của subscriber connection tăng (slow consumer).
• Streams: stream length vượt MAXLEN dự kiến; PEL size tăng (consumer không ack); consumer group lag (last-delivered-id cách xa cuối stream).

17. Tổng Kết & Quiz

Điểm cốt lõi

• Pub/Sub: at-most-once, fire-and-forget, throughput cao, latency thấp, subscriber offline mất message vĩnh viễn.
• Streams: at-least-once (với XACK), persist, consumer offline đọc lại từ cursor, consumer group cho parallel processing.
• Rule of thumb: mất message = hậu quả nghiêm trọng → Stream. Mất message chấp nhận được, cần fan-out → Pub/Sub.
• Hybrid pattern: ghi Stream trước (durability), PUBLISH sau (delivery). User online nhận Pub/Sub; user offline đọc Stream khi reconnect.
• Streams only + XREAD BLOCK: durability mà không cần hybrid, trade-off latency 1–5ms và throughput thấp hơn ~20–50%.
• Anti-pattern chính: Pub/Sub cho payment/order, Stream không MAXLEN, XREADGROUP không XACK.
• MAXLEN bắt buộc cho mọi Stream trong production.

Quiz

1. Một hệ thống gửi thông báo "like" cho user. User thường online. Nếu miss vài notification khi offline không ảnh hưởng nghiêm trọng. Bạn chọn Pub/Sub hay Stream? Lý do?
2. Hệ thống thanh toán cần đảm bảo mỗi payment event được xử lý đúng một lần dù worker crash. Nên dùng cơ chế nào? Liệt kê các lệnh Redis liên quan.
3. Với hybrid pattern (Stream + Pub/Sub), nếu PUBLISH thất bại sau khi XADD thành công, hệ thống xử lý thế nào để không mất data?
4. PEL là gì? Nó grow vô hạn trong trường hợp nào? Cách khắc phục?
5. User A đang online, user B đang offline. Cả hai đều subscribe chat room 123. Với hybrid pattern, mỗi user nhận tin nhắn mới như thế nào?

Đáp án gợi ý

1. Pub/Sub phù hợp: miss vài notification khi offline chấp nhận được, không cần persist. Pub/Sub latency thấp hơn, memory footprint thấp hơn, subscriber join/leave đơn giản hơn. Nếu cần user đọc lại notification history, thêm Stream hoặc DB riêng.
2. Stream + consumer group. Các lệnh: XADD (ghi event), XREADGROUP (đọc và assign), XACK (confirm xử lý), XCLAIM/XAUTOCLAIM (reassign khi worker crash). Kết hợp idempotency key để tránh double-process khi retry.
3. XADD đã lưu vào Stream — message an toàn. User offline sẽ đọc lại qua XREAD khi reconnect. User online miss Pub/Sub nhưng sẽ đọc được từ Stream nếu client có logic fallback (load_missed_messages). Không có data loss; chỉ có delivery delay cho user online.
4. PEL (Pending Entry List) là danh sách message đã được XREADGROUP deliver cho consumer nhưng chưa được XACK. Nó grow vô hạn khi consumer đọc message (XREADGROUP) nhưng không bao giờ gọi XACK (ví dụ consumer crash liên tục hoặc code quên XACK). Khắc phục: XACK ngay sau xử lý thành công; dùng XAUTOCLAIM để reassign message pending quá lâu (idle consumer); monitor PEL size qua XPENDING.
5. User A online: WebSocket server đang SUBSCRIBE channel "chat:channel:123" → nhận message qua Pub/Sub <1ms ngay khi tin nhắn được gửi. User B offline: không có connection nào subscribe channel → PUBLISH không deliver tới B. Khi B reconnect, client gọi load_missed_messages với last_read_id đã lưu → XRANGE Stream từ sau ID đó → nhận tất cả tin nhắn đã bỏ lỡ.

Bài tiếp theo

Bài 74 — Failure Recovery: xử lý khi Redis hoặc gateway sập trong hệ thống realtime, bao gồm reconnect strategy, message recovery, và graceful degradation.

Tham khảo

• Redis Docs — Pub/Sub
• Redis Docs — Streams
• Redis Docs — XADD
• Redis Docs — XREADGROUP
• Redis Docs — XAUTOCLAIM
• Redis Docs — XREAD
• Redis Docs — PUBLISH
• Redis Docs — SUBSCRIBE