Redis Pipeline 批量写入出错怎么办：逐条结果核对、重试边界与幂等键

订单导入任务一次往 Redis 写入几百个商品状态时，日志里最容易出现一种误判：客户端返回了错误，就把整批命令再发一遍。Pipeline 只负责把多条命令集中发送、集中读取结果，并不保证这一批要么全部成功、要么全部失败。正确的处理方式是逐条核对返回值，按命令类型决定是否重试，再用幂等键把重复副作用挡在业务边界之外。

先把 Pipeline 的成功边界说清楚

Redis Pipeline 的核心收益是把多次请求合并到较少的网络往返里。服务器仍然按顺序处理每条命令，客户端也会收到一组对应结果；其中一条命令报错，不会自动撤销前面已经完成的写入。

例如一批命令里先写入 product:42:stock，再给 product:42:updated_at 写入时间，第三条命令因为参数类型不对失败，前两条不会凭空消失。这个事实决定了故障处理的第一步：保存命令顺序和业务主键，逐项判断结果。

用 go-redis 做一次可观察的批量写入

下面的例子把每条命令绑定到一个业务记录。示例使用 github.com/redis/go-redis/v9，重点不在连接初始化，而在发送后如何读取结果。生产代码里还应把批次号写入日志，方便把 Redis 返回和导入文件中的行对应起来。

type WriteItem struct {
    Key   string
    State string
}

func writeStates(ctx context.Context, rdb *redis.Client, items []WriteItem) error {
    pipe := rdb.Pipeline()
    for _, item := range items {
        pipe.Set(ctx, item.Key, item.State, 10*time.Minute)
    }

    cmds, err := flushBatch(ctx, pipe)
    if err != nil {
    // 这里的 err 只能说明批次存在失败，不能代替逐条检查。
        log.Printf("redis batch has failures: %v", err)
    }

    var failed []string
    for i, cmd := range cmds {
        if cmd.Err() != nil {
            failed = append(failed, fmt.Sprintf("index=%d key=%s err=%v", i, items[i].Key, cmd.Err()))
        }
    }
    if len(failed) > 0 {
        return fmt.Errorf("batch partial failure: %s", strings.Join(failed, "; "))
    }
    return nil
}

这里的关键是 cmds[i] 与 items[i] 的位置关系。不要只把批量发送返回的总错误写入日志后就丢掉结果数组；也不要假设失败项一定集中在数组末尾。命令级错误可能来自错误类型、参数非法，也可能来自连接中断后的未知状态，它们的后续动作不同。

把失败分成“明确失败”和“结果未知”

排查时可以先按证据分两类。Redis 已经返回一个明确错误，例如对字符串执行哈希命令，这一条通常没有执行成功，可以记录下来并修正输入。另一类是客户端等待响应超时，连接在什么时候断开并不确定：命令可能还没到 Redis，也可能已经执行，只是响应没回来。

哪些写入可以重试，哪些必须先补幂等

覆盖式写入的重试边界相对清楚。例如把订单状态设置为 paid，重复发送通常仍然得到同一个最终值。但如果写入本身代表一次事件，重复执行就可能产生两次库存扣减、两条队列消息或两次积分增加。

一个简单的幂等做法是给每次业务动作分配 request_id，先用 done:{request_id} 记录处理状态，再执行副作用。需要注意，这两个动作如果仍然分成普通 Pipeline 命令，并不天然具备原子性；要把“检查并占用”做成真正的原子操作，必须使用 Redis 脚本或把状态收口到能提供条件写入的业务存储。

requestID := "order-9007-stock-20260724-01"
doneKey := "done:" + requestID

// 只展示重试判断，副作用动作不能只依赖客户端内存标记。
seen, err := rdb.Exists(ctx, doneKey).Result()
if err != nil {
    return err
}
if seen == 1 {
    return nil
}
// 后续应使用原子占用方案，再执行一次库存变更。

如果只是同步一批商品的“当前状态”，优先使用 SET/HSET 这类可覆盖结果，并在 value 里带版本号或更新时间。重试时比较版本，旧批次不能覆盖新批次。这里别急着把所有失败都塞进重试队列，先确认重复写入会不会改变业务事实。

重试实现要小，边界要能复查

实践中更稳妥的批处理通常是“小批次 + 单条失败清单 + 有上限的补发”。例如每 100 条命令组成一个批次；首次发送后只把明确可重试的 key 放入补发列表，最多补发两次，并把批次号、命令下标和最终状态写入日志。超时批次则先进入待确认状态，而不是直接判定失败。

• 批次大小：从 50～200 条起压测，观察 payload 大小和单次响应时间。
• 重试次数：一般设置硬上限，避免 Redis 故障时形成重试风暴。
• 退避时间：使用带随机抖动的短退避，多个 worker 不要同时补发。
• 最终状态：成功、明确失败、待确认三态分开存储，便于人工或定时任务处理。

复查时至少看三件事：失败清单是否能还原到原始输入行，补发后目标 key 的版本是否正确，以及重试次数是否在告警阈值内。只看“批次成功率”很容易漏掉少数关键订单。

常见问题：Redis Pipeline 重试怎么判断

Pipeline 失败会自动回滚吗？

不会。Pipeline 只是批量传输和读取结果，已经执行的命令不会因为后续命令失败而自动回滚。

网络超时后能不能把整批再发一次？

不能直接这么做。超时代表结果未知，应先查询幂等记录或目标状态；只有确认重复执行不会产生副作用时，才适合补发。

SET 为什么比 INCR 更适合直接重试？

SET 是把 key 写成指定值，重复执行通常得到相同结果；INCR 会在每次执行时继续累加，没有请求号或原子幂等控制就可能重复计数。

go-redis 的总错误能代替逐条结果吗？

不能。总错误只能提示批次存在异常，真正需要处理的是结果数组中每条命令的错误和对应业务 key。

把“批量成功”改成可验证的结果

Redis Pipeline 最适合减少大量小写入的网络成本，但它不会替业务决定一致性。让每条命令带上可追踪的业务键，保存逐条结果，把明确失败和未知结果分开，再为累加、追加类动作设计幂等占用，这套处理就能经得住断线和局部故障。最终验收不要只问“这批发出去了吗”，而要回答“哪些 key 已确认、哪些需要补查、重试是否改变了业务结果”。