Redis pipeline 怎样处理部分命令失败

Redis pipeline 出现部分命令失败时，应把每条回复对应回原命令，保留已确认的成功结果，单独处理命令错误；如果是超时或断连，未取得完整回复的操作可能已经执行，不能直接重放整批。pipeline 是批量发送手段，不是失败自动回滚的承诺。

官方地址：https://redis.readthedocs.io/en/stable/advanced_features.html

我更关心每条命令有没有明确结果

假设批量刷新缓存时，一条命令访问了错误的数据类型，另外两条却是正常写入。我更关心的是“每条操作有没有可信回复”，而不是给整个批次贴一个失败标签。把一个报错转换为整批重试，代码看起来省事，却容易让已经成功的递增、追加操作重复生效。

这里讨论同步 redis-py 的单实例、非事务 pipeline，不涉及 Cluster、WATCH 或脚本。我的取舍是：先把结果分成确认成功、确认命令错误、结果未知三类，再考虑重试。这个分类比“异常就再执行一次”多一点记录工作，但能让调用方知道接下来需要处理什么。

准备一个不会覆盖业务键的错误样本

为避免示例影响业务，使用独立本地测试实例和两个此前不存在的键。下面的完整 Python 示例假定已安装 redis-py，连接本机默认端口，不含真实凭据。它主动制造一个 WRONGTYPE：先把一个测试键写成字符串，再在 pipeline 中对它执行 LPUSH。

示例会检查测试键是否存在，并用 NX 防止覆盖；它仍不是共享生产环境的并发隔离机制，只适合无人并发写入这些键的独立实验环境。遇到已占用前缀就停止，不自动删除旧键。

关闭事务包装并逐条读取回复

redis-py 普通 Redis 客户端创建 pipeline 时默认启用事务包装，因而这里明确设置 transaction=False。execute(raise_on_error=False) 让普通命令的 ResponseError 留在返回列表里，便于逐条判断；它不会把网络异常也变成正常元素。

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry
from redis.exceptions import ResponseError, ConnectionError, TimeoutError

# 仅用于独立本地实验；明确关闭连接错误自动重试
client = redis.Redis(
    host="127.0.0.1", port=6379, decode_responses=True,
    socket_connect_timeout=2, socket_timeout=2,
    retry=Retry(NoBackoff(), 0),
)
wrong_key = "demo:pipe-partial:text"
counter_key = "demo:pipe-partial:count"

try:
    # 不覆盖已有键；发现冲突就停止实验
    if client.exists(wrong_key, counter_key):
        raise RuntimeError("测试键已存在，请使用新的独立前缀")
    if not client.set(wrong_key, "not-a-list", nx=True):
        raise RuntimeError("测试前缀被占用")
    if not client.set(counter_key, "0", nx=True):
        raise RuntimeError("计数样本被占用")

    # 操作标签与入队顺序一致，不记录敏感参数
    operations = ["递增计数", "向字符串追加列表项", "读取计数"]
    try:
        with client.pipeline(transaction=False) as pipe:
            pipe.incr(counter_key)
            pipe.lpush(wrong_key, "entry")
            pipe.get(counter_key)
            replies = pipe.execute(raise_on_error=False)
    except (ConnectionError, TimeoutError) as exc:
        # 缺少完整回复不等于没有执行，禁止在这里整批重放
        print("批次结果未知", type(exc).__name__)
    else:
        if len(replies) != len(operations):
            raise RuntimeError("回复数量不符，不能按成功处理")
        for index, (operation, reply) in enumerate(zip(operations, replies)):
            if isinstance(reply, ResponseError):
                print(index, operation, "命令错误", type(reply).__name__)
            else:
                print(index, operation, "正常回复", repr(reply))
finally:
    # 关闭客户端连接；不在异常清理中悄悄重发或删除状态
    client.close()

在新建样本、无并发写入且连接正常的前提下，预期第一条回复为整数 1，第二条为 ResponseError，第三条为字符串 '1'。这是待复现的预期，不是本文实测记录。重要的是中间命令错误没有使前面的递增消失，后面的读取也能获得自己的回复。默认抛出首个命令错误时，同样不能据此推断后续命令从未执行。

把连接异常留在未知结果区

最容易让我对自动重试保持谨慎的，是“服务端已做了事，客户端却没收到回复”的情况。比如 INCR 已生效，回复途中断连；再次 INCR 会产生新的递增。客户端是否能证明某条请求尚未发送，应以明确的接口与连接信息为依据；没有这种证据时，不能把未知偷换成未执行。

代码不仅没有应用层重试循环，还显式把客户端 Retry 次数设为 0。实际系统需要同时查看客户端、连接池、封装层和任务队列的重试策略，避免外层没有重试、底层却已重放请求。此处不建议复制一个适用于所有客户端版本的默认重试假设。

我的建议是把未知结果放进待核对状态，由业务操作标识和可信业务记录决定是否补做。仅凭当前计数值通常也不足以证明某次递增是否发生，因为其他请求可能同时修改它。若要用去重标识保证补做安全，需要把去重判断和业务修改设计为同一原子操作；先查一个标识再单独递增，并不能自动解决竞态。

让日志能对应命令而不是只有批次

日志至少保存批次标识、操作索引、业务操作标识、命令名、键的脱敏标识、回复类别和尝试次数。不要只留下“pipeline failed”，也不要把含敏感值的完整命令参数直接打印出来。示例只打印操作标签和错误类型，是为了把错误位置说清楚，而不是收集私密数据。

需要特别区分 ResponseError 与正常但不满足业务预期的回复。例如条件写入没有满足条件，可能返回正常的否定结果，而不是异常对象。实际接入时为每种命令定义成功条件，不能把“不是异常”一律等同于“业务目标已完成”。独立命令可以独立处理；如果后面的操作依赖前面的结果，不能假设 pipeline 会根据前一条失败自动跳过。

上线前划清自动重试的范围

上线前确认三件事：结果与命令索引一一对应；未知结果不会被整批自动重试；错误样本不会触及业务键。WRONGTYPE 通常需要修正键模型或命令选择，重复发送同一个错误命令没有帮助。即使是重复 SET 相同值，看似幂等，也要考虑并发更新、过期时间或覆盖新值的影响，不能只按命令名决定重试安全。

如果需要事务，请单独设计事务语义，不要仅把 transaction 改成 True 就认为有数据库式回滚。Redis 事务中的执行期错误也不提供整体回滚；入队期错误和执行期错误的行为不同，不能把本文非事务示例的分支原样当成事务恢复方案。

独立实验结束后，先确认以下两个键确为本次样本，再执行删除；删除不可撤销，不要换成业务键或通配符：

# 仅清理自己创建的两个实验键，不使用 FLUSHDB 或通配符
redis-cli DEL demo:pipe-partial:text demo:pipe-partial:count

raise_on_error=False 是否会忽略所有异常？

不会。它控制普通命令错误的返回方式，不代表连接错误、超时或所有客户端异常都能变成回复元素。

失败命令修好后能否单独补发？

先确认它确实返回命令错误，再检查它与其他操作的依赖、当前数据和重试副作用；只有满足业务前提才补发，不机械地抽取所有失败项重试。

哪些任务适合这样使用 pipeline？

我更倾向于把它用于可独立解释结果、批次规模受控的操作。依赖上一条查询结果才能构造下一条命令的任务，或必须有完整业务原子性的写入，应该另选明确的交互或原子机制。

参考：redis-py pipeline 用法、客户端错误处理实现、Retry 配置、Redis 事务与错误。