《Redis 管道、事务与批量优化：从 N 次 RTT 到一次》

# 同机房 RTT ~0.5ms，服务端处理 ~0.05ms
# 跨地域 RTT ~30ms，服务端处理仍只有 ~0.05ms
# 距离越远，RTT 开销越致命

# 执行 100 条独立命令串行等待：总耗时 ≈ 100 × (RTT + 处理)
# 同机房 0.5ms RTT → ≈55ms；跨地域 30ms RTT → ≈3 秒

# 观察网络延迟
redis-cli -h 10.0.0.1 -p 6379 --latency
# min: 0, max: 5, avg: 0.52 (1000 samples)

# -P 16 表示 pipeline 16，吞吐可提升数十倍
redis-benchmark -h 10.0.0.1 -p 6379 -n 100000 -c 50 -P 16

# 普通模式：SET→回复→SET→回复...（N 次往返）
SET a 1
OK
SET b 2
OK

# pipeline 模式：一次性发送多条，一次性收回复
SET a 1
SET b 2
SET c 3
# 回复依次返回: OK OK OK

// Go 示例：go-redis Pipelined，一次往返执行全部
pipe := rdb.Pipeline()
pipe.Set(ctx, "a", 1, 0)
pipe.Set(ctx, "b", 2, 0)
pipe.Set(ctx, "c", 3, 0)
cmds, _ := pipe.Exec(ctx)

// Java 示例：Lettuce 异步批处理
RedisAsyncCommands<String, String> async = conn.async();
async.set("a", "1"); async.set("b", "2"); async.set("c", "3");

# redis-cli 也支持 --pipe 批量导入（每行一条命令）
cat commands.txt | redis-cli --pipe

# 未使用 pipeline：约 5 万次/秒（RTT 受限）；pipeline 16：约 200 万次/秒
redis-benchmark -n 100000 -c 50 -P 16

MULTI                     # 开启事务
SET balance:alice 100
DECRBY balance:alice 30
INCRBY balance:bob 30
EXEC                      # 顺序执行所有命令
# 返回: [OK, 85, 30]

MULTI
SET key1 value1
DISCARD                   # 取消事务，清空队列
# 返回 OK

WATCH balance:alice        # 监控余额
GET balance:alice          # 100
MULTI
DECRBY balance:alice 30
INCRBY balance:bob 30
EXEC
# 若期间 alice 余额被改 → EXEC 返回 nil，需重试
# 若未被改 → 正常执行，返回 [OK, 85, 30]

# 应用层乐观重试（伪代码）
while true:
    WATCH key
    val = GET key
    MULTI; 执行写逻辑; result = EXEC
    if result == nil: continue   # 冲突 → 重试
    break                        # 成功

# 纯批量读（如缓存预热、批量 MGET）→ pipeline
# 批量写但可容忍中间穿插 → pipeline
# 批量写且要求原子 → MULTI/EXEC 或 Lua
# 需要条件判断的原子操作 → Lua
# 需要事务内逻辑（如扣库存判断）→ Lua

# 错误：把事务当 pipeline 用，只为省 RTT
MULTI
GET key1      # 不需要原子性，却付出事务的队列开销
GET key2
EXEC

# 正确：纯读用 pipeline 或 MGET
MGET key1 key2

# 场景：导入 100 万用户缓存，原始做法 500 万次 SET × 每次 RTT，耗时极长
SET user:1:name tom
SET user:1:age 18
# ...

// 优化：pipeline 批量写，1 万条一次往返
pipe := rdb.Pipeline()
for i := 0; i < 10000; i++ {
    pipe.HSet(ctx, fmt.Sprintf("user:%d", i), "name", names[i], "age", ages[i])
}
pipe.Exec(ctx)

# 过大 → 客户端内存高、阻塞服务端太久；过小 → 收益不明显
# 经验值：单批 100~1000 条，视命令复杂度和带宽而定

-- 原子扣库存：pipeline/事务无法在服务端做条件判断
local stock = tonumber(redis.call('GET', KEYS[1]))
local qty = tonumber(ARGV[1])
if stock == nil then return -1 end
if stock >= qty then
    redis.call('DECRBY', KEYS[1], qty)
    return stock - qty
else
    return -2
end

EVAL "$(cat deduct.lua)" 1 stock:sku:001 5

# 高频 Lua 脚本务必用 EVALSHA 省去脚本体传输
SCRIPT LOAD "return redis.call('GET', KEYS[1])"
EVALSHA <sha1> 1 mykey

# 同一批 1000 条 SET：pipeline 0.09ms/条，MULTI/EXEC 0.12ms/条，Lua 0.10ms/条
# 三者差距不大，决策应由语义需求驱动

# 场景：批量写入 + 部分失败补偿
# 方案：pipeline 分批写 + 对失败批次用事务重试

// 分批 pipeline 通用写法：每批 500，Exec 后检查错误并补偿
for batch := 0; batch < total; batch += 500 {
    pipe := rdb.Pipeline()
    // 填充 500 条命令
    cmds, _ := pipe.Exec(ctx)
}

# 用 INFO stats 的 total_commands_processed 对比优化前后吞吐斜率
redis-cli info stats
# 用 SLOWLOG 观察大批量命令是否拖慢
redis-cli slowlog get 10

# 集群中 pipeline 只能包含同槽 key
# 跨槽批量需按槽分组，或使用 hash tag
MGET {user:1}:name {user:1}:age
# 同一槽 → 允许

# 1. pipeline 不保证原子，中途可被其他命令插入
# 2. MULTI/EXEC 无回滚，运行时错误不中断后续命令
# 3. Lua 脚本阻塞主线程，须保持短小
# 4. WATCH 冲突重试要有上限，防止活锁
# 5. 大批量命令避免在高峰期执行；集群中 pipeline 限同槽 key