Redis Pipeline 批处理：游标持久化的高性能实践

背景

在实际业务场景中，我们经常需要批量处理大量 Redis 数据。例如：

批量导入数据到 Redis
批量查询大量 key
批量更新缓存

传统的逐条命令执行方式存在严重的性能瓶颈：每条命令都需要经过网络往返（RTT），当数据量大时，网络延迟会成为主要瓶颈。

Pipeline 原理

Redis Pipeline 是一种批量执行命令的优化技术，核心思想是：

将多条命令打包发送到服务端
服务端依次执行所有命令
将所有结果打包返回给客户端

这样可以大幅减少网络往返次数，提升吞吐量。

基础用法

Go 语言示例（go-redis）

go
package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
)

func pipelineBasic(ctx context.Context, rdb *redis.Client) error {
    // 创建 Pipeline
    pipe := rdb.Pipeline()
    
    // 批量添加命令
    for i := 0; i < 1000; i++ {
        key := fmt.Sprintf("key:%d", i)
        value := fmt.Sprintf("value:%d", i)
        pipe.Set(ctx, key, value, 0)
    }
    
    // 一次性执行所有命令
    _, err := pipe.Exec(ctx)
    return err
}

Python 示例（redis-py）

python
import redis

def pipeline_basic(r: redis.Redis):
    # 创建 Pipeline
    pipe = r.pipeline()
    
    # 批量添加命令
    for i in range(1000):
        key = f"key:{i}"
        value = f"value:{i}"
        pipe.set(key, value)
    
    # 一次性执行所有命令
    pipe.execute()

性能对比

以批量写入 10000 条数据为例：

方式	耗时	说明
逐条执行	~10s	每条命令一次网络往返
Pipeline	~0.1s	一次网络往返执行所有命令

性能提升约 100 倍！

游标持久化实践

在处理超大数据集时（如百万级），一次性 Pipeline 可能导致：

内存溢出
阻塞时间过长
失败后无法断点续传

分批 Pipeline + 游标持久化

go
// 批量处理大数据集，支持断点续传
func BatchProcessWithCursor(ctx context.Context, rdb *redis.Client, data []DataItem, batchSize int, cursorKey string) error {
    // 1. 获取上次处理的游标位置
    cursor, _ := rdb.Get(ctx, cursorKey).Int()
    if cursor >= len(data) {
        return nil // 已处理完成
    }
    
    // 2. 从游标位置开始分批处理
    for i := cursor; i < len(data); i += batchSize {
        end := i + batchSize
        if end > len(data) {
            end = len(data)
        }
        
        // 3. 使用 Pipeline 批量处理当前批次
        pipe := rdb.Pipeline()
        for j := i; j < end; j++ {
            item := data[j]
            pipe.Set(ctx, item.Key, item.Value, 0)
        }
        
        // 4. 更新游标位置（原子操作）
        pipe.Set(ctx, cursorKey, end, 0)
        
        if _, err := pipe.Exec(ctx); err != nil {
            return fmt.Errorf("batch %d-%d failed: %w", i, end, err)
        }
        
        // 5. 记录进度
        log.Printf("Processed %d/%d items", end, len(data))
    }
    
    return nil
}

关键设计要点

游标存储: 将处理进度存储在 Redis 中，支持断点续传
原子更新: 将数据操作和游标更新放在同一个 Pipeline 中，保证一致性
合理批次大小: 根据数据大小和网络状况调整，建议 100-1000 条/批
错误处理: 记录失败批次，支持重试

实战案例：批量迁移缓存

go
// 从旧缓存批量迁移到新缓存
func MigrateCache(ctx context.Context, oldRdb, newRdb *redis.Client, pattern string, batchSize int) error {
    var cursor uint64
    processed := 0
    
    for {
        // SCAN 扫描旧缓存
        keys, nextCursor, err := oldRdb.Scan(ctx, cursor, pattern, int64(batchSize)).Result()
        if err != nil {
            return err
        }
        
        if len(keys) > 0 {
            // Pipeline 批量读取
            readPipe := oldRdb.Pipeline()
            cmds := make([]*redis.StringCmd, len(keys))
            for i, key := range keys {
                cmds[i] = readPipe.Get(ctx, key)
            }
            
            if _, err := readPipe.Exec(ctx); err != nil && err != redis.Nil {
                return err
            }
            
            // Pipeline 批量写入新缓存
            writePipe := newRdb.Pipeline()
            for i, cmd := range cmds {
                val, err := cmd.Result()
                if err == nil {
                    writePipe.Set(ctx, keys[i], val, 0)
                }
            }
            
            if _, err := writePipe.Exec(ctx); err != nil {
                return err
            }
        }
        
        processed += len(keys)
        cursor = nextCursor
        
        if cursor == 0 {
            break
        }
    }
    
    log.Printf("Migration completed: %d keys", processed)
    return nil
}

注意事项

1. 内存控制

Pipeline 会将所有命令缓存到内存中，批量过大可能导致 OOM：

go
// ❌ 错误：一次性处理太多数据
pipe := rdb.Pipeline()
for i := 0; i < 1000000; i++ {  // 百万条命令
    pipe.Set(ctx, key, value, 0)
}
pipe.Exec(ctx)

// ✅ 正确：分批处理
for i := 0; i < 1000000; i += 1000 {
    pipe := rdb.Pipeline()
    for j := i; j < i+1000 && j < 1000000; j++ {
        pipe.Set(ctx, key, value, 0)
    }
    pipe.Exec(ctx)
}

2. 原子性保证

Pipeline 不保证原子性，如果需要原子操作，使用 MULTI/EXEC：

go
// 需要原子性时使用 Transaction
_, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
    pipe.Set(ctx, "key1", "value1", 0)
    pipe.Set(ctx, "key2", "value2", 0)
    return nil
})

3. 超时设置

大批量操作可能超时，需要调整超时时间：

go
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

最佳实践总结

合理分批: 每批 100-1000 条命令
游标持久化: 存储处理进度，支持断点续传
原子更新: 数据操作与游标更新同 Pipeline
错误重试: 记录失败批次，支持重试机制
监控进度: 日志记录处理进度
资源控制: 控制并发数，避免压垮 Redis

总结

Redis Pipeline 是提升批量操作性能的利器，结合游标持久化可以实现：

高性能: 减少网络往返，吞吐量提升 100 倍
可靠性: 断点续传，失败重试
可扩展: 支持百万级数据处理

在处理大规模数据时，Pipeline + 游标持久化是一种成熟可靠的解决方案。

目录