Bài 68: Sharded Pub/Sub (Redis 7) cho Cluster

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

• Hiểu tại sao PUBLISH/SUBSCRIBE scale kém trong Redis Cluster và overhead cụ thể là gì.
• Nắm cơ chế hash slot routing của SPUBLISH/SSUBSCRIBE.
• Dùng được hash tag để group nhiều channel về cùng node.
• Viết publisher và subscriber bằng redis-py 4.x với cluster client.
• Biết khi nào migrate từ regular sang sharded, khi nào không cần.

2. Vấn Đề: Pub/Sub Thường Trong Cluster

Bài 65 đã giải thích cơ chế cơ bản của PUBLISH/SUBSCRIBE. Bài này tập trung vào hành vi của nó trong Cluster mode và vấn đề phát sinh khi scale.

Cluster Bus và broadcast

Redis Cluster có một kênh giao tiếp nội bộ gọi là cluster bus (port +10000 so với data port). Mỗi node duy trì kết nối full-mesh với tất cả node khác qua bus này để đồng bộ trạng thái cluster (gossip, failover, slot mapping).

Khi bạn gọi PUBLISH channel message vào bất kỳ node nào, Redis Cluster sẽ forward message đó tới tất cả node trong cluster qua cluster bus, vì bất kỳ node nào cũng có thể đang có subscriber lắng nghe channel đó. Đây là hành vi có chủ đích: đảm bảo không có subscriber nào bị bỏ sót.

Overhead khi cluster lớn

Hệ quả trực tiếp: với cluster N node, mỗi lần PUBLISH một message, message đó được truyền N lần (1 lần đến node gốc, N-1 lần qua cluster bus đến các node còn lại). Thực tế còn phức tạp hơn vì cluster bus dùng TCP full-mesh.

• Cluster 3 node: 1 PUBLISH → 3 lần truyền — overhead nhỏ, chấp nhận được.
• Cluster 10 node: 1 PUBLISH → 10 lần truyền — bắt đầu cảm nhận.
• Cluster 50 node: 1 PUBLISH → 50 lần truyền — throughput Pub/Sub thực tế giảm mạnh.

Kết quả: thêm node để scale data capacity lại khiến Pub/Sub throughput giảm — nghịch chiều với mục tiêu scale-out. Cluster lớn nhưng Pub/Sub traffic cao trở thành điểm nghẽn.

Ví dụ đo lường

Giả sử mỗi node xử lý được 100.000 msg/s Pub/Sub. Với regular Pub/Sub và cluster 10 node:

• Tổng capacity cluster: vẫn ~100.000 msg/s (vì mọi node đều phải xử lý mọi message).
• Thêm node không tăng được throughput, chỉ tăng cluster bus traffic.

Với Sharded Pub/Sub và cluster 10 node:

• Mỗi channel chỉ đi qua 1 node → 10 channel song song trên 10 node khác nhau → tổng ~1.000.000 msg/s.
• Thêm node = thêm capacity thực sự.

3. Sharded Pub/Sub — Giải Pháp Redis 7.0

Redis 7.0 (phát hành tháng 4/2022) giới thiệu Sharded Pub/Sub với hai lệnh chính: SPUBLISH và SSUBSCRIBE. Tư tưởng cốt lõi: thay vì broadcast message tới toàn cluster, message chỉ tồn tại trong một hash slot — và một hash slot thuộc về đúng một node.

Nguyên lý hoạt động

1. Channel name được hash theo thuật toán CRC16 → ra hash slot (0–16383).
2. Hash slot đó thuộc về một node cụ thể (slot owner).
3. SPUBLISH channel message: cluster client tự route đến đúng node owner của slot, message được deliver cho subscriber trên node đó.
4. SSUBSCRIBE channel: cluster client kết nối đến đúng node owner của slot để lắng nghe.
5. Không có cluster bus broadcast — node khác không biết message này tồn tại.

Tại sao scale tốt hơn

Mỗi channel (hay nhóm channel cùng hash slot) được xử lý bởi một node độc lập. Capacity Pub/Sub tổng cộng bằng tổng capacity của tất cả node. Thêm node vào cluster → thêm capacity Pub/Sub — đúng với mục tiêu scale-out.

4. Hash Slot Routing

Redis Cluster chia không gian key thành 16384 hash slot (0–16383). Slot của một key được tính:

slot = CRC16(key) % 16384

Với Sharded Pub/Sub, channel name được xử lý như key: cùng thuật toán CRC16, cùng modulo 16384.

Ví dụ thực tế

# Giả sử cluster 3 node:
# Node A: slot 0–5460
# Node B: slot 5461–10922
# Node C: slot 10923–16383

# CRC16("chat:room:abc") % 16384 = 7638  → Node B
# CRC16("chat:room:xyz") % 16384 = 11203 → Node C
# CRC16("notifications")  % 16384 = 3821  → Node A

# SPUBLISH "chat:room:abc" "Hello" → cluster client tự redirect đến Node B
# SSUBSCRIBE "chat:room:abc"       → cluster client kết nối Node B để lắng nghe

MOVED redirect

Nếu client connect nhầm node, Redis trả về MOVED slot ip:port — giống hành vi với key thông thường. Cluster client (redis-py ClusterPubSub, ioredis cluster) xử lý MOVED tự động và reconnect đến đúng node. Bạn không cần can thiệp thủ công.

5. Hash Tag Để Group Channel

Nếu channel name chứa cặp ngoặc nhọn {...}, Redis chỉ hash phần bên trong dấu ngoặc để tính slot. Đây là hash tag — cơ chế tương tự với key thông thường trong Cluster.

Vì sao cần hash tag cho Pub/Sub

Xét bài toán chat room: một room có nhiều loại channel — tin nhắn, trạng thái gõ, thông báo vào/ra. Bạn muốn tất cả channel của cùng một room nằm trên cùng một node để giảm network hop khi subscriber trong phòng phải lắng nghe nhiều channel.

# Không có hash tag — 3 channel của cùng room có thể nằm 3 node khác nhau:
# "chat:msg:room123"    → CRC16("chat:msg:room123")    → slot X → Node A
# "chat:typing:room123" → CRC16("chat:typing:room123") → slot Y → Node B
# "chat:presence:room123" → CRC16("chat:presence:room123") → slot Z → Node C

# Với hash tag {} — 3 channel cùng về node chứa slot của "room123":
# "chat:msg:{room123}"      → CRC16("room123") → slot S → Node B
# "chat:typing:{room123}"   → CRC16("room123") → slot S → Node B
# "chat:presence:{room123}" → CRC16("room123") → slot S → Node B

Subscriber của room 123 chỉ cần kết nối đến Node B, không cần duy trì connection với 3 node khác nhau.

Chú ý

Hash tag khiến nhiều channel dồn vào cùng một slot → cùng một node. Nếu số lượng room ít mà traffic mỗi room rất cao, có thể tạo hotspot. Thiết kế channel name sao cho số lượng entity (room_id, user_id) đủ lớn và phân tán đều.

6. Commands Mới: SPUBLISH, SSUBSCRIBE, SUNSUBSCRIBE

SPUBLISH

SPUBLISH channel message

Publish message vào channel sharded. Chỉ subscriber đang SSUBSCRIBE channel đó (trên cùng node) nhận được. Trả về số lượng subscriber nhận được message.

Lưu ý: subscriber dùng SUBSCRIBE (regular) sẽ không nhận được message từ SPUBLISH. Hai hệ thống hoàn toàn tách biệt về semantics.

SSUBSCRIBE / SUNSUBSCRIBE

SSUBSCRIBE channel [channel ...]
SUNSUBSCRIBE channel [channel ...]

Subscribe/unsubscribe các sharded channel. Connection vào subscriber mode — chỉ chấp nhận SSUBSCRIBE, SUNSUBSCRIBE, PING, QUIT sau đó (giống SUBSCRIBE thường). Message nhận về có type "smessage" thay vì "message".

PUBSUB SHARDCHANNELS và SHARDNUMSUB

# Liệt kê sharded channel đang có subscriber (trên node hiện tại)
PUBSUB SHARDCHANNELS [pattern]

# Đếm số subscriber per channel (trên node hiện tại)
PUBSUB SHARDNUMSUB [channel ...]

Hai lệnh này trả về thông tin của node hiện tại, không aggregate toàn cluster. Để biết toàn cảnh, bạn cần query từng node.

Pattern subscription không được hỗ trợ

Không có SPSUBSCRIBE. Sharded Pub/Sub chỉ hỗ trợ exact channel name — vì pattern không thể map về một hash slot cụ thể. Đây là limitation by design, không phải bug. Nếu cần pattern, dùng regular Pub/Sub với PSUBSCRIBE.

7. Code Python redis-py 4.x

redis-py hỗ trợ Sharded Pub/Sub từ version 4.4.0 thông qua RedisCluster và phương thức pubsub() trả về ClusterPubSub.

Publisher

import redis.cluster

# Kết nối cluster — truyền bất kỳ node nào trong cluster
r = redis.cluster.RedisCluster(
    host="localhost",
    port=7000,
    decode_responses=True,
)

# SPUBLISH: trả về số subscriber nhận được message
count = r.spublish("chat:room:{abc}", "Hello room abc")
print(f"Delivered to {count} subscriber(s)")

# Thêm message
r.spublish("chat:room:{abc}", "Second message")
r.spublish("chat:room:{xyz}", "Message for room xyz")  # node khác với {abc}

Subscriber

import redis.cluster
import time

r = redis.cluster.RedisCluster(
    host="localhost",
    port=7000,
    decode_responses=True,
)

# pubsub() của RedisCluster trả về ClusterPubSub
pubsub = r.pubsub()

# SSUBSCRIBE — cluster client route đến đúng node của slot
pubsub.ssubscribe("chat:room:{abc}")

print("Listening for messages...")
for msg in pubsub.listen():
    # Sharded message có type "smessage" (khác "message" của regular)
    if msg["type"] == "smessage":
        print(f"Channel: {msg['channel']}, Data: {msg['data']}")
    elif msg["type"] == "ssubscribe":
        print(f"Subscribed to {msg['channel']}, subscriber count: {msg['data']}")

Subscribe nhiều channel cùng node

# Nhiều channel cùng hash tag → cùng node → 1 connection xử lý
pubsub.ssubscribe(
    "chat:msg:{room123}",
    "chat:typing:{room123}",
    "chat:presence:{room123}",
)

for msg in pubsub.listen():
    if msg["type"] == "smessage":
        channel = msg["channel"]
        data = msg["data"]
        if channel.startswith("chat:msg:"):
            handle_new_message(data)
        elif channel.startswith("chat:typing:"):
            handle_typing_indicator(data)
        elif channel.startswith("chat:presence:"):
            handle_presence_update(data)

Non-blocking với get_message

pubsub.ssubscribe("notif:{user_42}")

# Polling trong event loop — không block
while True:
    msg = pubsub.get_message(ignore_subscribe_messages=True)
    if msg and msg["type"] == "smessage":
        process(msg["data"])
    time.sleep(0.001)  # 1ms poll

8. Use Case 1 — Chat Room Theo Shard

Chat là use case phù hợp nhất với Sharded Pub/Sub: message của một room không cần broadcast sang room khác, và scope tự nhiên là per-room.

import json
import redis.cluster

r = redis.cluster.RedisCluster(host="localhost", port=7000, decode_responses=True)

def send_message(room_id: str, user_id: str, text: str):
    """Publish tin nhắn vào room."""
    payload = json.dumps({"user": user_id, "text": text})
    # Hash tag {room_id} đảm bảo cùng room → cùng node
    channel = f"chat:room:{{{room_id}}}"
    count = r.spublish(channel, payload)
    return count

def listen_room(room_id: str):
    """Subscribe và xử lý message của room."""
    channel = f"chat:room:{{{room_id}}}"
    pubsub = r.pubsub()
    pubsub.ssubscribe(channel)
    for msg in pubsub.listen():
        if msg["type"] == "smessage":
            data = json.loads(msg["data"])
            print(f"[{room_id}] {data['user']}: {data['text']}")

# Publisher
send_message("room_101", "alice", "Hello everyone")
send_message("room_101", "bob", "Hi Alice!")

# Subscriber (chạy trong thread/process riêng)
listen_room("room_101")

Toàn bộ traffic của room_101 xử lý trên một node duy nhất. Node đó không gửi bất kỳ thứ gì qua cluster bus cho hai node còn lại.

9. Use Case 2 — Notification Per User

Notification trực tiếp tới user là pattern phù hợp khác: mỗi user có channel riêng, scope hẹp.

import json
import redis.cluster

r = redis.cluster.RedisCluster(host="localhost", port=7000, decode_responses=True)

def push_notification(user_id: str, notification: dict):
    """Gửi notification realtime đến user đang online."""
    channel = f"notif:{{{user_id}}}"
    r.spublish(channel, json.dumps(notification))

def listen_notifications(user_id: str):
    """User instance subscribe channel của mình."""
    channel = f"notif:{{{user_id}}}"
    pubsub = r.pubsub()
    pubsub.ssubscribe(channel)
    for msg in pubsub.listen():
        if msg["type"] == "smessage":
            notif = json.loads(msg["data"])
            print(f"Notification for {user_id}: {notif}")

# Gửi notification từ backend
push_notification("user_42", {"type": "message", "from": "user_7", "preview": "Hey!"})
push_notification("user_42", {"type": "like", "post_id": "post_99"})

# User's WebSocket handler subscribe khi user kết nối
listen_notifications("user_42")

Với 1 triệu user, traffic notification phân tán đều trên tất cả node theo hash slot của user_id. Không node nào là bottleneck.

10. Regular vs Sharded — Bảng So Sánh

11. Khi Nào Dùng, Khi Nào Không

Nên dùng Sharded Pub/Sub khi

• Cluster từ 5 node trở lên với Pub/Sub traffic cao.
• Channel có scope tự nhiên theo entity: room, user, session, tenant.
• Đang gặp bottleneck cluster bus do PUBLISH thường.
• Cần scale Pub/Sub throughput theo số node.

Không cần Sharded Pub/Sub khi

• Chạy single Redis instance — không có cluster bus, PUBLISH/SUBSCRIBE là đủ.
• Cluster nhỏ (2–3 node) với Pub/Sub traffic thấp — overhead cluster bus không đáng kể.
• Cần broadcast global: thông báo admin, config reload, invalidate cache trên toàn cluster. Regular Pub/Sub là đúng tool cho trường hợp này.
• Cần PSUBSCRIBE pattern matching — Sharded không hỗ trợ.
• Library chưa hỗ trợ sharded (redis-py < 4.4, ioredis chưa stable sharded API).

12. Migration Từ Regular Sang Sharded

Migration không phải drop-in replacement vì semantics khác nhau. Cần ba bước:

Bước 1 — Cập nhật library

# Kiểm tra version hiện tại
pip show redis
# Cần redis-py >= 4.4.0 để dùng ClusterPubSub với ssubscribe
pip install "redis[hiredis]>=4.4.0"

Bước 2 — Cập nhật channel naming

# Trước: channel không có hash tag
old_channel = f"chat:room:{room_id}"       # → slot khác nhau với typing channel
old_channel = f"chat:typing:{room_id}"

# Sau: thêm hash tag để group theo room_id
new_channel = f"chat:room:{{{room_id}}}"    # → cùng slot với typing
new_channel = f"chat:typing:{{{room_id}}}"

Bước 3 — Thay lệnh

# Publisher:
# r.publish(channel, message)    # trước
# r.spublish(channel, message)   # sau

# Subscriber:
# pubsub.subscribe(channel)      # trước → type "message"
# pubsub.ssubscribe(channel)     # sau  → type "smessage"

# Xử lý message — cập nhật type check:
for msg in pubsub.listen():
    # if msg["type"] == "message":      # trước
    if msg["type"] == "smessage":       # sau
        handle(msg["data"])

Triển khai an toàn

Nếu hệ thống cần zero-downtime, chạy song song hai hệ thống: publisher gửi cả PUBLISH và SPUBLISH trong thời gian transition, subscriber dần cắt sang SSUBSCRIBE. Khi tất cả subscriber đã chuyển, tắt PUBLISH.

13. Hybrid Pattern

Không phải chọn một trong hai. Nhiều ứng dụng dùng cả hai loại trong cùng một hệ thống:

import redis.cluster

r = redis.cluster.RedisCluster(host="localhost", port=7000, decode_responses=True)

# --- Regular Pub/Sub: global broadcast ---
# Thông báo admin, maintenance mode, feature flag update
# Cần mọi app server nhận → PUBLISH broadcast toàn cluster là đúng
def broadcast_maintenance(message: str):
    r.publish("system:maintenance", message)

def broadcast_feature_flag(flag: str, enabled: bool):
    r.publish("system:feature_flags", f"{flag}:{int(enabled)}")

# --- Sharded Pub/Sub: per-entity realtime ---
# Chat, notification — không cần cross-room/cross-user broadcast
def send_chat(room_id: str, payload: str):
    r.spublish(f"chat:{{{room_id}}}", payload)

def send_notification(user_id: str, payload: str):
    r.spublish(f"notif:{{{user_id}}}", payload)


# Subscriber cũng dùng cả hai:
regular_pubsub = r.pubsub()
regular_pubsub.subscribe("system:maintenance", "system:feature_flags")

sharded_pubsub = r.pubsub()
sharded_pubsub.ssubscribe(f"chat:{{{room_id}}}", f"notif:{{{user_id}}}")

Rule of thumb: scope global → regular; scope per-entity → sharded.

14. Limitations & Anti-patterns

Limitations cần nhớ

• Không có SPSUBSCRIBE: pattern subscription không hỗ trợ. Nếu code hiện tại phụ thuộc vào PSUBSCRIBE, không thể migrate channel đó sang sharded.
• Chỉ hoạt động trong Cluster mode: gọi SPUBLISH/SSUBSCRIBE trên single instance hoặc Sentinel sẽ báo lỗi. Cần abstract trong code để fallback.
• Connection model khác: mỗi subscriber duy trì connection tới node owner của slot. Nếu shard nhiều loại channel cho nhiều node, subscriber có thể cần nhiều connection hơn.
• Library support: không phải mọi client đều hỗ trợ ổn. Kiểm tra changelog trước khi update.

Anti-patterns

• PUBLISH thường trong Cluster lớn với high traffic: đây là vấn đề bài này giải quyết — nếu gặp bottleneck, migrate sang SPUBLISH.
• Channel name không hash tag khi muốn group: "chat:msg:room1" và "chat:typing:room1" sẽ nằm khác node. Dùng {room1} để collocate.
• Mix SPUBLISH với SUBSCRIBE: subscriber dùng SUBSCRIBE không nhận được message từ SPUBLISH — hoàn toàn tách biệt. Debug rất khó nếu nhầm.
• SPUBLISH trong single Redis instance: không có cluster → không có slot owner → lỗi. Cần detect cluster mode hoặc dùng try/except để fallback về PUBLISH.
• Quá nhiều channel cùng hash tag vào một slot: tạo hotspot. Đảm bảo key space phân tán đều (ví dụ: số lượng room_id đủ lớn).

Fallback detect cluster mode

def smart_publish(r, channel: str, message: str):
    """Publish sharded nếu cluster, regular nếu single instance."""
    try:
        # Kiểm tra cluster info
        info = r.cluster_info()
        if info.get("cluster_enabled") == 1:
            return r.spublish(channel, message)
    except Exception:
        pass
    return r.publish(channel, message)

15. Tổng Kết & Quiz

Điểm cốt lõi

• Regular Pub/Sub trong Cluster: PUBLISH broadcast qua cluster bus tới tất cả node → O(N) overhead, throughput không scale.
• Sharded Pub/Sub (Redis 7.0+): SPUBLISH/SSUBSCRIBE route theo hash slot → message chỉ ở 1 node → O(1) cluster overhead, throughput scale theo node.
• Hash tag {} trong channel name để group nhiều channel về cùng slot/node.
• Message type nhận về là "smessage", không phải "message".
• Không có SPSUBSCRIBE — pattern subscription không hỗ trợ trong sharded.
• Hybrid hợp lý: regular cho global broadcast, sharded cho per-entity.

Quiz

1. Với Redis Cluster 20 node, mỗi lần gọi PUBLISH channel msg thì message đó đi qua bao nhiêu node? Với SPUBLISH thì sao?
2. Channel "chat:typing:room99" và "chat:msg:room99" có đảm bảo cùng node trong Sharded Pub/Sub không? Sửa lại để đảm bảo.
3. Một subscriber dùng SUBSCRIBE "events". Publisher gọi SPUBLISH "events" "hello". Subscriber có nhận được message không? Giải thích.
4. Ứng dụng chat có feature "thông báo toàn server khi bảo trì". Nên dùng PUBLISH hay SPUBLISH? Tại sao?
5. Kể hai trường hợp không nên migrate từ regular sang sharded Pub/Sub.

Đáp án gợi ý

1. Với PUBLISH: 20 node — message broadcast qua cluster bus đến tất cả node. Với SPUBLISH: 1 node — chỉ node owner của hash slot nhận và deliver.
2. Không đảm bảo — CRC16 của hai channel name khác nhau có thể ra hai slot khác nhau, nằm hai node khác nhau. Sửa: "chat:typing:{room99}" và "chat:msg:{room99}" — cả hai đều hash theo "room99" → cùng slot → cùng node.
3. Không. Regular SUBSCRIBE và Sharded SPUBLISH là hai hệ thống tách biệt hoàn toàn. SPUBLISH chỉ deliver đến subscriber dùng SSUBSCRIBE. SUBSCRIBE không nhận được gì từ SPUBLISH.
4. Nên dùng PUBLISH (regular). Thông báo bảo trì cần broadcast toàn cluster để mọi app server instance đều nhận — đây là use case cốt lõi của regular Pub/Sub. SPUBLISH chỉ deliver đến subscriber của một slot/node, sẽ bỏ sót phần lớn app server.
5. (1) Single Redis instance hoặc Sentinel — không có Cluster mode, SPUBLISH báo lỗi. (2) Cần PSUBSCRIBE pattern matching — sharded không hỗ trợ, phải giữ regular.

Bài tiếp theo

Bài 69 — Presence Tracking: theo dõi ai đang online trong hệ thống realtime, dùng kết hợp Sorted Set, Pub/Sub, và TTL.