Redis 发布订阅与流解决的是不同的投递问题。发布订阅面向当前生效的订阅广播，不保留供以后回放的消息；流保存条目，读者可以在它们仍被保留时重新访问。选择哪种机制，首先要回答谁必须收到事件，以及消费者缺席期间应该发生什么。

先确定每个事件交给谁

对于后续状态可以替代的实时看板更新，只向在线用户广播也许已经足够。但在假设新听众已经到场之前，应等待订阅确认。重连恢复的是未来投递，不是错过的发布。PUBLISH 返回正数，表示按 Redis 命令语义统计到了接收者，不等于应用确认业务处理已经完成。

流有两种不同的消费方式。普通 XREAD 让独立读者分别指定自己的起始 ID，双方可以收到同一份保留历史。它既不自动分担工作，也不替每个连接保管检查点。同一个消费组中的 XREADGROUP 则在具名消费者之间分配新投递，并在未启用 NOACK 时记录待处理条目。

不同消费组允许账单、分析等独立工作流在同一条流上分别推进。组内增加成员是在分担工作，新建一组则是增加一套投递历史。消费者名字不同、工作进程变多，都不承诺负载均分或业务按序完成。应先确定需要的是广播、独立回放还是竞争处理，再选择对应的读取命令，不能把三种模式混为一谈。

保留窗口不是持久化设置

XADD 把数据保存在 Redis 里，并不自动保证每个已接受的写入都能熬过崩溃或故障切换。RDB/AOF 设置、复制行为、键淘汰及恢复操作共同决定持久性。发布订阅也不会因为启用了持久化，就突然拥有可回放的日志。因此，把两者概括成时效性与持久性的取舍，会掩盖各自的真实契约。

MAXLEN 删除的是最旧的条目，不是裁掉最新队尾、同时让早期历史一直保留。近似裁剪可能留下额外记录，精确裁剪则遵循指定数量；两者都不提供无限回放。默认裁剪还可能删除待处理引用对应的载荷，因此应围绕允许的最长停机时间与恢复后的追赶过程设计保留策略和容量。

保留积压也有运行成本。需要控制增长，分别观察未读滞后和待处理任务，并定义无法正常处理的消息如何处置。发布订阅虽然不保存历史，却仍使用连接输出缓冲；慢听众触达相应限制后可能被断开。两种机制都不能让内存规划消失，只是占用与丢失发生的位置不同。

坏掉的样子

用发布订阅派发必须执行的任务，订阅者缺席时就会失去投递机会。能够重新加载权威状态的看板可能接受这点，必须执行的财务操作通常不能接受。合适的替代品可以是保留消息的流，也可以是其他持久队列，但恢复语义必须围绕业务操作设计，不能从产品名字直接推导出来。

流若没有恢复循环，会出现另一种失败：已经投递但未确认的条目停留在待处理列表，普通 > 读取却继续处理新消息。处理前确认可能丢工作，处理后确认则允许重试时重复副作用。如果重复执行不可接受，稳定的事件标识和幂等处理仍是必需的，不能仅靠建组和追加消息解决。

把同一事件分别 PUBLISH 和 XADD，也不是原子双写协议。执行若在两次操作之间中断，可能只有一边成功。下面的实验在隔离进程中有意比较两种机制，不是在推荐用这一串命令构成生产送达保证。比较实验里的执行顺序，不应被直接提升为跨系统一致性方案。

快速验证

原有实验记录使用 macOS、Python 3.14.4、Redis 8.10.1 和 redis-py 8.1.0；本轮审阅没有重新执行服务实验，下面的结果说明保留原有观察范围。准备 POSIX 环境、C 编译器、make、curl、tar 与 Python 3.10 或更新版本。安装步骤仅用 libc 构建 Redis 核心，并把客户端装入临时虚拟环境。

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

在同一终端用 "$redis_lab/venv/bin/python" 运行下面的 Python 代码。程序在 /tmp 下创建私有 Unix socket，关闭 TCP、定时快照和追加日志，最后只终止自己启动的进程并清理运行目录。下载依赖需要网络，不会连接默认端口上的本机 Redis。

from contextlib import contextmanager
import os
from pathlib import Path
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry


@contextmanager
def instance():
    with TemporaryDirectory(prefix="redis-lab-", dir="/tmp") as directory:
        endpoint = str(Path(directory) / "redis.sock")
        process = subprocess.Popen([
            os.environ["REDIS_SERVER"], "--port", "0",
            "--unixsocket", endpoint, "--unixsocketperm", "700",
            "--save", "", "--appendonly", "no", "--dir", directory,
            "--maxmemory-policy", "noeviction",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        client = redis.Redis(unix_socket_path=endpoint, decode_responses=False,
                             socket_timeout=2, retry=Retry(NoBackoff(), 0))
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "Redis exited during startup"
                try:
                    if client.ping():
                        break
                except redis.ConnectionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.02)
            yield client
        finally:
            client.close()
            if process.poll() is None:
                process.terminate()
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors.decode()


def receive(subscription):
    deadline = time.monotonic() + 2
    while time.monotonic() < deadline:
        response = subscription.get_message(timeout=0.1)
        if response is not None:
            return response
    raise AssertionError("subscription response timed out")


with instance() as client:
    assert client.publish("updates", b"offline") == 0
    first_id = client.xadd("history", {"event": "offline"}, id="1-0")
    with client.pubsub() as first, client.pubsub() as second:
        for subscription in (first, second):
            subscription.subscribe("updates")
            assert receive(subscription)["type"] == "subscribe"
        assert client.publish("updates", b"live") == 2
        second_id = client.xadd("history", {"event": "live"}, id="2-0")
        for subscription in (first, second):
            response = receive(subscription)
            assert response["type"] == "message" and response["data"] == b"live"
    with client.client() as reader_one, client.client() as reader_two:
        one = reader_one.xread({"history": "0-0"})
        two = reader_two.xread({"history": "0-0"})
    assert one == two
    assert [entry[0] for entry in one[0][1]] == [first_id, second_id]
    print("Pub/Sub listeners get live only; both independent XREAD readers replay offline and live")

    client.xgroup_create("history", "workers", id="0-0")
    a = client.xreadgroup("workers", "a", {"history": ">"}, count=1)[0][1]
    b = client.xreadgroup("workers", "b", {"history": ">"}, count=1)[0][1]
    assert a[0][0] == first_id and b[0][0] == second_id
    assert client.xpending("history", "workers")["pending"] == 2
    assert client.xack("history", "workers", first_id, second_id) == 2
    assert client.xlen("history") == 2
    print("one group splits new entries between a/b; ACK clears pending state, not stored history")

    third_id = client.xadd("history", {"event": "later"}, id="3-0")
    assert client.xtrim("history", maxlen=2, approximate=False) == 1
    retained = client.xread({"history": "0-0"})[0][1]
    assert [entry[0] for entry in retained] == [second_id, third_id]
    assert client.xrange("history", first_id, first_id) == []
    print("exact MAXLEN removes the oldest entry; replay is limited to retained history")

没有听众时发布 offline 返回零，而 XADD 保留了同名事件条目。等两个订阅都确认后，每个听众收到的是 live，而非 offline。随后两个普通 XREAD 都返回两条已保存记录，说明它们在独立回放，不是在把工作分成互不重叠的两份。

创建一个消费组后，投递方式发生变化：a、b 两个消费者得到不同条目，二者都被记入待处理列表。确认它们不会改变流长度。追加第三条后，精确 MAXLEN 只删除最旧条目；此时即使从零开始读取，也找不回它已经被删除的载荷，回放能力受保留窗口约束。

三条输出摘要都来自对真实响应的断言，展示了订阅在场、独立回放、组内分配与有限保留。实验没有比较延迟和内存，没有复现真实崩溃，没有验证持久化或自动恢复。这些运行决策仍然属于应用的投递契约，不能用一次本机实验成功替代。

来源