【redis】pipeline管道

Redis Pipeline（管道）是一种将多个命令批量发送到服务器的技术。与逐个发送命令的传统方式不同，Pipeline允许客户端一次性打包多条命令，通过单次网络往返完成批量操作。

为什么需要Pipeline？

在常规操作模式下，每个Redis命令的执行流程为：

客户端发送命令 -> 服务器处理 -> 客户端接收响应

当执行N条命令时会产生N次网络往返延迟（RTT）。

而Pipeline通过批量发送机制：

客户端打包命令1,2,3... -> 服务器顺序处理 -> 批量返回响应

将网络开销从O(N)降低到O(1)，在批量操作场景下可带来数十倍的性能提升。

适用场景：

• 批量写入/读取数据

• 需要执行多个无关联命令

• 高并发且对实时性要求不高的操作

Pipeline命令的使用

命令行使用

使用redis-cli的--pipe参数实现Pipeline：

# 生成测试命令数据（生成5条SET命令）
$ printf "SET key%d value%d\n" $(seq 1 10) > commands.txt

# 查询生成的测试命令数据
$ cat commands.txt
SET key1 value2
SET key3 value4
SET key5 value6
SET key7 value8
SET key9 value10

# 通过Pipeline执行
$ cat commands.txt | redis-cli --pipe
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 5

Java代码实现

使用Jedis客户端库示例：

package com.morris.redis.demo.pipeline;


import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.util.List;
import java.util.concurrent.ExecutionException;

/**
 * jedis使用pipeline
 */
public class JedisPipelineDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Jedis jedis = new Jedis("localhost", 6379);
        Pipeline pipeline = jedis.pipelined();

        // 批量写入
        for (int i = 1; i <= 10000; i++) {
            pipeline.set("key" + i, "value" + i);
        }

        // 提交并获取响应（注意响应顺序与命令顺序一致）
        List<Object> responses = pipeline.syncAndReturnAll();
        System.out.println("写入完成，响应数量：" + responses.size());
    }
}

性能对比测试

测试代码：

package com.morris.redis.demo.pipeline;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

/**
 * pipeline和单命令执行性能对比
 */
public class PerformanceTestDemo {

    public static final int count = 1000;

    public static void main(String[] args) {
        // 普通模式
        long start = System.currentTimeMillis();
        try (Jedis jedis = new Jedis("localhost", 6379)) {
            for (int i = 0; i < count; i++) {
                jedis.set("normal_" + i, "value");
            }
        }
        long normalCost = System.currentTimeMillis() - start;

        // Pipeline模式
        start = System.currentTimeMillis();
        try (Jedis jedis = new Jedis("localhost", 6379)) {
            Pipeline pipeline = jedis.pipelined();
            for (int i = 0; i < count; i++) {
                pipeline.set("pipe_" + i, "value");
            }
            pipeline.sync();
        }
        long pipeCost = System.currentTimeMillis() - start;

        System.out.printf("操作次数: %d | 普通模式: %dms | Pipeline模式: %dms\n",
                count, normalCost, pipeCost);
    }
}

测试结果：

操作次数: 1000 | 普通模式: 2784ms | Pipeline模式: 2140ms
操作次数: 10000 | 普通模式: 8315ms | Pipeline模式: 185ms
操作次数: 100000 | 普通模式: 52180ms | Pipeline模式: 2479ms

注意事项

1. 批量大小控制：建议单次Pipeline命令数控制在10,000以内

2. 非原子性：Pipeline不是事务，中间命令失败不会影响后续执行

3. 内存消耗：服务端会将所有响应缓存在内存中，超大Pipeline可能引发OOM

4. 错误处理：通过解析响应列表检查每个命令的执行结果