如何获取Redis频道订阅前已发布的消息？

这个问题我之前做消息服务的时候也碰到过，核心原因很明确：Redis原生的Pub/Sub是“实时推送”模式，不会持久化任何历史消息，只有在你成功订阅频道之后发布的消息，才会被推送给客户端。要拿到启动前已经存在的消息，必须换用支持消息持久化的Redis数据结构，或者在发布环节额外做持久化处理。下面给你几个实用的方案：

方案1：改用Redis Stream（最推荐） #

Redis Stream是官方专门为持久化消息队列设计的特性，它会把所有消息存在流里，支持按消息ID/偏移量读取历史内容，同时也能实时监听新消息，完全覆盖你的需求。

给你改一下你的示例代码：

import redis
import threading
import time
import settings

def msg_handler():
    r = redis.client.StrictRedis(host='localhost', port=6379, db=0)
    # 首次启动从最早的消息开始读（用"0"表示起始ID）
    # 如果需要从上次程序退出的位置继续，把last_id存在本地文件/Redis里就行
    last_id = "0"
    
    # 先读取所有历史消息
    while True:
        # count=0表示读取所有未处理的历史消息，block=0表示阻塞等待（如果还有的话）
        messages = r.xread({settings.REDIS_CHANNEL: last_id}, count=0, block=0)
        for stream, msgs in messages:
            for msg_id, msg_data in msgs:
                print(f"历史消息: {msg_data}")
                last_id = msg_id  # 更新到最后一条已处理的消息ID
        # 历史消息读完后，持续监听新消息
        new_messages = r.xread({settings.REDIS_CHANNEL: last_id}, block=0)
        for stream, msgs in new_messages:
            for msg_id, msg_data in msgs:
                print(f"新消息: {msg_data}")
                last_id = msg_id

def main():
    for i in range(3):
        t = threading.Thread(target=msg_handler, name=f'worker-{i}')
        print(f'thread {i}')
        t.setDaemon(True)
        t.start()
    while True:
        print('Waiting')
        time.sleep(1)

注意：发布消息的时候要改用xadd命令，比如：

# 发布消息到Stream
r.xadd(settings.REDIS_CHANNEL, {"content": "你的消息内容"})

Stream还支持分组消费、消息确认等高级特性，适合生产环境的复杂场景。

方案2：用Redis List模拟消息队列 #

如果不想改太多代码，也可以用List来做持久化：

• 发布消息时用lpush把消息塞进List
• 订阅端启动时，先把List里的历史消息全部读完，再用brpop阻塞等待新消息

示例代码：

def msg_handler():
    r = redis.client.StrictRedis(host='localhost', port=6379, db=0)
    # 先读取所有历史消息
    while True:
        msg = r.rpop(settings.REDIS_CHANNEL)
        if not msg:
            break  # 历史消息读完了
        print(f"历史消息: {msg}")
    # 持续监听新消息（brpop会阻塞直到有新消息）
    while True:
        msg = r.brpop(settings.REDIS_CHANNEL, timeout=0)
        if msg:
            print(f"新消息: {msg[1]}")

发布消息的代码改成：

r.lpush(settings.REDIS_CHANNEL, "你的消息内容")

这个方案的缺点是消息被读取后会从List里删除，如果有多个消费者需要获取同一份历史消息，就得给每个消费者维护自己的偏移量，或者用其他方式复制消息。

方案3：Pub/Sub结合持久化（兼容原有发布逻辑） #

如果你的生产环境已经在用Pub/Sub发布消息，不想修改发布端的代码，可以在发布消息的同时，把消息备份到一个List里，订阅端启动时先读List里的历史，再切换到Pub/Sub听新消息。

发布端修改（加个备份逻辑）：

def publish_msg(channel, msg):
    r = redis.client.StrictRedis(host='localhost', port=6379, db=0)
    # 原有的Pub/Sub发布
    r.publish(channel, msg)
    # 同时把消息存入历史List
    history_key = f"{channel}_history"
    r.lpush(history_key, msg)
    # 可选：限制历史消息数量，避免占太多内存，比如只保留最近1000条
    r.ltrim(history_key, 0, 999)

订阅端修改：

def msg_handler():
    r = redis.client.StrictRedis(host='localhost', port=6379, db=0)
    history_key = f"{settings.REDIS_CHANNEL}_history"
    
    # 先读取所有历史消息
    while True:
        msg = r.rpop(history_key)
        if not msg:
            break
        print(f"历史消息: {msg}")
    
    # 再订阅实时消息
    sub = r.pubsub()
    sub.subscribe(settings.REDIS_CHANNEL)
    while True:
        msg = sub.get_message()
        if msg and msg['type'] == 'message':
            print(f"新消息: {msg['data']}")

这个方案可以无缝兼容原有发布逻辑，但要注意维护历史消息的存储和清理，避免内存溢出。

最后总结 #

• 如果你是新项目，优先选Redis Stream，它是官方推荐的持久化消息队列方案，功能完善，适合生产环境。
• 如果要兼容现有Pub/Sub系统，用Pub/Sub+List备份的方式最省心，但要做好历史消息的内存管控。

内容的提问来源于stack exchange，提问作者ddd