发布订阅与 Stream

发布订阅（Pub/Sub）

Pub/Sub 是 Redis 内置的消息广播机制。发布者（Publisher）向频道（Channel）发送消息，所有订阅该频道的订阅者（Subscriber）同时收到消息，实现解耦通信。

Pub/Sub 命令速查表

Pub/Sub Python 示例

import redis, threading, time

r = redis.Redis(host='localhost', decode_responses=True)

# 订阅者（在独立线程中运行）
def subscriber(channels: list):
    pubsub = r.pubsub()
    pubsub.subscribe(*channels)
    print(f"订阅频道: {channels}")
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"收到消息 [{message['channel']}]: {message['data']}")

# 模式订阅（订阅所有 news.* 频道）
def pattern_subscriber():
    pubsub = r.pubsub()
    pubsub.psubscribe('news.*')
    for message in pubsub.listen():
        if message['type'] == 'pmessage':
            print(f"模式匹配 [{message['channel']}]: {message['data']}")

# 启动订阅者线程
t1 = threading.Thread(target=subscriber, args=(['news.tech', 'news.sports'],), daemon=True)
t2 = threading.Thread(target=pattern_subscriber, daemon=True)
t1.start(); t2.start()
time.sleep(0.1)

# 发布消息
r.publish('news.tech', 'Redis 7.2 发布了！')
r.publish('news.sports', '世界杯决赛即将开始')

Stream：Redis 的持久化消息流

Redis 5.0 引入的 Stream 是一种仿 Kafka 设计的持久化日志数据结构，支持：消息持久化、消费者组（同一消息只被一个消费者处理）、待确认消息（PEL）、消息回溯。

Stream 命令速查表

消费者组完整示例（Python）

import redis, json, time, threading

r = redis.Redis(host='localhost', decode_responses=True)
STREAM_KEY = "stream:orders"
GROUP_NAME = "payment-service"

# 初始化：创建 Stream 和消费者组
def init_stream():
    try:
        # $ 表示只消费从现在开始的新消息
        r.xgroup_create(STREAM_KEY, GROUP_NAME, id='$', mkstream=True)
        print(f"消费者组 {GROUP_NAME} 创建成功")
    except redis.exceptions.ResponseError as e:
        if 'BUSYGROUP' in str(e):
            print(f"消费者组 {GROUP_NAME} 已存在")

# 生产者：发布订单消息
def produce_order(order: dict) -> str:
    msg_id = r.xadd(STREAM_KEY, order,
                       maxlen=10000)  # 自动裁剪，保留最新 10000 条
    print(f"[Producer] 消息ID: {msg_id}, 订单: {order}")
    return msg_id

# 消费者：消费者组模式
def consume_orders(consumer_name: str):
    print(f"[{consumer_name}] 开始消费")
    while True:
        try:
            # XREADGROUP: 读取未投递的新消息，阻塞 2 秒
            messages = r.xreadgroup(
                GROUP_NAME, consumer_name,
                {STREAM_KEY: '>'},
                count=10,
                block=2000
            )
            if not messages:
                continue

            for stream, msgs in messages:
                for msg_id, data in msgs:
                    try:
                        process_order(data)
                        # 成功处理后 ACK（从待确认列表移除）
                        r.xack(STREAM_KEY, GROUP_NAME, msg_id)
                        print(f"[{consumer_name}] ACK: {msg_id}")
                    except Exception as e:
                        # 处理失败：消息留在 PEL，等待重试或转移
                        print(f"[{consumer_name}] 处理失败 {msg_id}: {e}")
        except Exception as e:
            print(f"[{consumer_name}] 错误: {e}")
            time.sleep(1)

def process_order(data: dict):
    time.sleep(0.05)  # 模拟支付处理

# 故障恢复：转移超时未确认消息
def recover_pending(timeout_ms: int = 30000):
    """将超过 30 秒未确认的消息转移给其他消费者"""
    pending = r.xpending_range(
        STREAM_KEY, GROUP_NAME,
        min='-', max='+', count=10,
        idle=timeout_ms
    )
    for msg in pending:
        r.xclaim(STREAM_KEY, GROUP_NAME,
                  'recovery-consumer', timeout_ms, msg['message_id'])

# 运行
init_stream()
for i in ['consumer-1', 'consumer-2']:
    threading.Thread(target=consume_orders, args=(i,), daemon=True).start()
for i in range(10):
    produce_order({'order_id': str(i), 'amount': str(i * 10)})

Stream vs Kafka：适用场景对比

Pub/Sub vs Stream：深度对比

Redis 7.x Sharded Pub/Sub

在 Redis Cluster 模式下，传统 Pub/Sub 的 PUBLISH 消息会广播到所有节点（每个节点都要处理），带来巨大的网络开销。Redis 7.0 引入的 Sharded Pub/Sub 将频道绑定到特定槽（slot），消息只在负责该槽的节点上传播：

# Sharded Pub/Sub（仅在 Cluster 模式下有意义）
# 发布：SPUBLISH 替代 PUBLISH
SPUBLISH mychannel "message"

# 订阅：SSUBSCRIBE 替代 SUBSCRIBE
SSUBSCRIBE mychannel

# Python 示例
from redis.cluster import RedisCluster

rc = RedisCluster(host='localhost', port=7001, decode_responses=True)

# 发布到分片频道（只路由到该频道所在槽的节点）
rc.spublish('notifications:user:1001', '{"type":"msg","content":"你好"}')

# 订阅（自动连接到负责该频道槽的节点）
sharded_pubsub = rc.pubsub()
sharded_pubsub.ssubscribe('notifications:user:1001')