---
tags: [redis, pipeline, batch, performance, multi-exec]
create time: 2026-06-01 10:30
---

# Pipeline 批量操作

## 概述

Redis 的每次命令执行都有网络往返（RTT）开销。对于需要执行多条命令的场景，**Pipeline** 可以把多个命令打包成一个批次发送，大幅降低 RTT 次数。这是限流性能优化的核心手段之一（F14 考点）。

## 一、为什么需要 Pipeline？

### 无 Pipeline：逐条发送

```
应用                          Redis
 │ ───INCR key1─────────────▶ │   (RTT #1)
 │ ◀──1────────────────────── │
 │                             │
 │ ───EXPIRE key1 60─────────▶ │   (RTT #2)
 │ ◀──1────────────────────── │
 │                             │
 │ ───INCR key2─────────────▶ │   (RTT #3)
 │ ◀──1────────────────────── │
 │                             │
 │ ───EXPIRE key2 60─────────▶ │   (RTT #4)
 │ ◀──1────────────────────── │
 │                             │
总 RTT: 4 次，延迟 = 4 × network_latency
```

### 有 Pipeline：批量发送

```
应用                          Redis
 │ ───INCR key1                │
 │ ───EXPIRE key1 60           │  (一次 TCP 发送)
 │ ───INCR key2                │
 │ ───EXPIRE key2 60─────────▶ │   (RTT #1)
 │ ◀──[1, 1, 1, 1]─────────── │
 │                             │
总 RTT: 1 次，延迟 = 1 × network_latency
```

**性能提升：** 如果网络 RTT 是 1ms，4 条命令从 4ms 降到 1ms——**节省了 75% 的网络延迟**。

## 二、Pipeline vs MULTI/EXEC

这是最容易混淆的概念。填空题 F14 的答案是 MULTI/EXEC，但实际使用中有重要的区别。

### 对比表

| 维度 | Pipeline | MULTI/EXEC |
|------|----------|------------|
| **原子性** | ❌ 每条命令独立执行 | ✅ EXEC 时整体执行 |
| **取消支持** | ❌ 不能中途取消 | ✅ DISCARD 取消 |
| **事务回滚** | ❌ 单条失败不影响其他 | ⚠️ EXEC 失败则全部不执行 |
| **嵌套管道** | ❌ 不能在事务内用 Pipeline | ❌ 不支持嵌套 |
| **Watch 支持** | ❌ 无 | ✅ 配合 WATCH 实现乐观锁 |

### 关键区别：原子性

```go
// Pipeline：每条命令独立执行，前一条成功后面失败也照常返回
pipe := rdb.Pipeline()
pipe.Incr(ctx, "key1")    // 成功
pipe.Expire(ctx, "key1", 60) // 即使这步出错，Incr 结果仍会返回
results, _ := pipe.Exec(ctx)

// MULTI/EXEC：EXEC 时所有命令作为一个整体执行
pipe2 := rdb.TxPipeline()  // TxPipeline = MULTI/EXEC 包装
pipe2.Incr(ctx, "key1")
pipe2.Expire(ctx, "key1", 60)
results2, _ := pipe2.Exec(ctx)
```

> [!tip]- Go-Redis 的 API 区分
>
> | API | 对应行为 |
> |-----|---------|
> | `rdb.Pipeline()` | 普通 Pipeline（非原子批量） |
> | `rdb.TxPipeline()` | MULTI/EXEC 事务包装（原子批量） |
>
> **生产环境推荐用 `TxPipeline()`**，因为它保证了操作的原子性。

## 三、在限流中的应用

### 3.1 多级限流的 Pipeline 优化

```go
// 原始方式：三次独立的 Redis 调用
func multiLevelLimitSlow(ctx context.Context, rdb *redis.Client, ip, userID, endpoint string) error {
    if err := checkIPLimit(ctx, rdb, ip); err != nil {
        return err // L1 拦截
    }
    if err := checkUserLimit(ctx, rdb, userID); err != nil {
        return err // L2 拦截
    }
    if err := checkTokenBucket(ctx, rdb, endpoint); err != nil {
        return err // L3 拦截
    }
    return nil
}

// Pipeline 优化：L2 + L3 合并为一个批次
func multiLevelLimitFast(ctx context.Context, rdb *redis.Client, ip, userID, endpoint string) error {
    // L1 单独调用（不可合并到同一个 Key space）
    if err := checkIPLimit(ctx, rdb, ip); err != nil {
        return err
    }
    
    // L2 + L3 用 Pipeline 打包
    pipe := rdb.TxPipeline()
    
    userResult := pipe.Eval(ctx, slidingWindowLua, []string{fmt.Sprintf("rate:user:{%s}", userID)}, 86400, 1000)
    tokenResult := pipe.Eval(ctx, tokenBucketLua, []string{fmt.Sprintf("rate:tokenbucket:{%s}", endpoint)}, 100, 10, 1, time.Now().UnixMilli())
    
    _, err := pipe.Exec(ctx)
    if err != nil {
        return err
    }
    
    // 检查结果
    userCount, _ := userResult.Int()
    if userCount >= 1000 {
        return ErrRateLimited
    }
    
    tokenResult, _ := tokenResult.IntSlice()
    if tokenResult[0] == 0 {
        return ErrRateLimited
    }
    
    return nil
}
```

### 3.2 ZAdd + EXPIRE 的 Pipeline 优化

```go
// 不用 Lua 时的简化写法（牺牲部分原子性换取简单）
func SlidingWindowPipeline(ctx context.Context, rdb *redis.Client, id string, windowSec int, maxLen int64) error {
    key := fmt.Sprintf("rate:sliding:%s", id)
    now := time.Now().UnixMilli()
    cutoff := now - int64(windowSec)*1000
    member := ulid.Now().String()

    pipe := rdb.TxPipeline()
    pipe.ZRemRangeByScore(ctx, key, "-inf", strconv.FormatInt(cutoff, 10))
    pipe.ZCard(ctx, key)
    results, err := pipe.Exec(ctx)
    if err != nil {
        return err
    }
    
    currentCount := int(results[1].(*redis.IntCmd).Val())
    if currentCount >= int(maxLen) {
        return ErrRateLimited
    }
    
    pipe2 := rdb.TxPipeline()
    pipe2.ZAdd(ctx, key, &redis.Z{Score: float64(now), Member: member})
    pipe2.Expire(ctx, key, time.Duration(windowSec)*time.Second)
    _, err = pipe2.Exec(ctx)
    return err
}
```

## 四、Pipeline 的注意事项

### 4.1 注意事项汇总

```mermaid
flowchart TD
    subgraph "⚠️ Pipeline 陷阱"
        T1["不能用 Pipeline 做条件逻辑<br/>无法根据 A 的结果决定 B"] --> T2["需要判断分支时用 Lua 脚本"]
        
        T3["大批量发送可能超过 Redis<br/>maxpacket 限制"] --> T4["控制单次 Pipeline 的命令数<br/>建议 50~200 条"]
        
        T5["流水线中的错误不会中断后续命令"] --> T6["需要在客户端检查每个命令的返回"]
        
        T7["Cluster 模式下多 Key<br/>必须在同一 slot"] --> T8["用 Hash Tag {} 保证同 slot"]
    end
    
    style T2 fill:#e3f2fd
    style T4 fill:#fff3e0
    style T6 fill:#fff3e0
    style T8 fill:#e3f2fd
```

### 4.2 最佳实践

| 场景 | 推荐方式 | 原因 |
|------|---------|------|
| 多条独立写入 | Pipeline (`TxPipeline`) | 减少 RTT，保持简单 |
| 读 → 判断 → 写 | Lua 脚本 | 需要原子性和业务逻辑 |
| 批量删除大 Key | UNLINK (逐个) | DEL 在大 Key 时会阻塞 |
| 统计类聚合操作 | Lua 中遍历 | 避免多次往返的数据不一致 |

## 五、性能基准参考

假设网络 RTT = 1ms，本地部署（RTT ≈ 0.1ms）：

| 命令数 | 无 Pipeline (RTT×N) | Pipeline (1 RTT) | 加速比 |
|--------|-------------------|-----------------|-------|
| 10 条 | 10ms / 1ms | 1ms / 0.1ms | **10x** |
| 100 条 | 100ms / 10ms | 1ms / 0.1ms | **100x** |
| 1000 条 | 1000ms / 100ms | 1ms / 0.1ms | **1000x** |

> [!note]- 实际应用
>
> 对于限流这种高频场景（每秒数千到数万请求），即使是 1ms 的额外 RTT 也可能成为瓶颈。Pipeline 的价值在于将 N 次 RTT 压缩为 1 次。

## 关联笔记

- [[分布式限流]] — 性能优化清单中的 Pipeline 章节
- [[Go-Redis Lua 调用指南]] — 与 EvalSHA 配合使用的组合策略
- [[SCAN命令]] — SCAN 遍历时可以结合 Pipeline 提高批量处理效率