Redis Pipeline与Lua脚本：批量操作性能优化的正确方式

Redis Pipeline与Lua脚本批量操作性能优化的正确方式同样是对 Redis 执行 1000 次 INCR 操作逐个发送需要 1000 次网络往返用 Pipeline 只需要 1 次。但如果你以为 Pipeline 就是批量操作的终点那可能还没有见识过 Lua 脚本在服务端原子执行时的威力。一、Redis 批量操作的三种方式与性能差距在对 Redis 做批量操作时有且仅有三种路径逐条执行、Pipeline 批量发送、Lua 脚本服务端执行。三者在网络开销、原子性保证、使用复杂度上存在本质差异。sequenceDiagram participant C as Redis Client participant N as Network participant R as Redis Server Note over C,R: 方式一逐条执行N次RTT loop 1000次 C-N: INCR counter:1 N-R: INCR counter:1 R-N: OK N-C: OK end Note over C,R: 方式二Pipeline1次RTT C-N: INCR counter:1br/INCR counter:2br/...br/INCR counter:1000 N-R: 1000条命令一次性发送 R-N: 1000条响应批量返回 N-C: 所有响应 Note over C,R: 方式三Lua脚本1次RTT 原子性 C-N: EVAL for i1,1000 dobr/redis.call(INCR,KEYS[i])br/end 1000 key1...key1000 N-R: 脚本参数 R-R: 服务端原子执行 R-N: 批量结果 N-C: 所有结果三者在 1000 次 INCR 操作场景下的实测数据本地 Redis网络延迟 ~0.1ms方式耗时相对逐条网络往返原子性逐条执行~105ms1x1000次无Pipeline~1.2ms87x1次无Lua 脚本~0.8ms131x1次有差距的根本原因不在 Redis 本身——Redis 单线程处理 1000 条命令只需要微秒级别——而在于网络往返次数RTT。逐条执行把 99% 的时间花在了网络上。二、Pipeline 的 RTT 优化原理与使用限制2.1 Pipeline 的工作原理Pipeline 不是 Redis 的服务端特性而是一个客户端特性。它的核心机制是客户端将多条命令打包在一个 TCP 包中发送然后一次性读取所有响应。Redis 服务端完全无感知——它看到的仍然是 1000 条独立的命令只是它们同时到达了而已。/** * Redis Pipeline 批量操作工具类 * 封装 Pipeline 的创建、执行、异常处理 */ Component Slf4j public class RedisPipelineTemplate { private final StringRedisTemplate stringRedisTemplate; private final RedisConnectionFactory connectionFactory; // Pipeline 单批次最大命令数防止单次发送数据量过大 private static final int MAX_BATCH_SIZE 500; public RedisPipelineTemplate(StringRedisTemplate stringRedisTemplate, RedisConnectionFactory connectionFactory) { this.stringRedisTemplate stringRedisTemplate; this.connectionFactory connectionFactory; } /** * 批量设置缓存自动分批 * * param kvMap 键值对映射 * param ttlSeconds 过期时间秒 * return 成功设置的数量 */ public int batchSet(MapString, String kvMap, long ttlSeconds) { if (kvMap null || kvMap.isEmpty()) { return 0; } int successCount 0; ListMap.EntryString, String entries new ArrayList(kvMap.entrySet()); // 分批执行 Pipeline避免单次发送过大 for (int i 0; i entries.size(); i MAX_BATCH_SIZE) { int end Math.min(i MAX_BATCH_SIZE, entries.size()); ListMap.EntryString, String batch entries.subList(i, end); successCount executeSetBatch(batch, ttlSeconds); } return successCount; } private int executeSetBatch(ListMap.EntryString, String batch, long ttlSeconds) { RedisConnection connection null; try { // 获取底层连接 connection RedisConnectionUtils.getConnection(connectionFactory); // 打开 Pipeline 模式 connection.openPipeline(); // 批量发送 SET 命令 for (Map.EntryString, String entry : batch) { byte[] key entry.getKey().getBytes(StandardCharsets.UTF_8); byte[] value entry.getValue().getBytes(StandardCharsets.UTF_8); if (ttlSeconds 0) { connection.setEx(key, ttlSeconds, value); } else { connection.set(key, value); } } // 执行并获取所有结果 ListObject results connection.closePipeline(); // 统计成功数量 int successCount 0; for (int i 0; i results.size(); i) { Object result results.get(i); // SET 成功返回 OK if (OK.equals(result)) { successCount; } else { log.warn(Pipeline SET 失败: key{}, result{}, batch.get(i).getKey(), result); } } return successCount; } catch (RedisException e) { log.error(Pipeline 批量SET异常: batchSize{}, batch.size(), e); throw new PipelineExecutionException(Pipeline 执行失败, e); } finally { if (connection ! null) { connection.close(); } } } /** * Pipeline 混合操作在一个 Pipeline 中执行多种命令 * 适用于需要批量获取不同类型数据的场景 */ public MapString, Object mixedPipeline(ListPipelineCommand commands) { if (commands null || commands.isEmpty()) { return Collections.emptyMap(); } RedisConnection connection null; try { connection RedisConnectionUtils.getConnection(connectionFactory); connection.openPipeline(); for (PipelineCommand cmd : commands) { byte[] key cmd.getKey().getBytes(StandardCharsets.UTF_8); switch (cmd.getType()) { case GET - connection.get(key); case HGETALL - connection.hGetAll(key); case ZCARD - connection.zCard(key); case LLEN - connection.lLen(key); case EXISTS - connection.exists(key); case TTL - connection.ttl(key); default - throw new IllegalArgumentException( 不支持的Pipeline命令类型: cmd.getType()); } } ListObject results connection.closePipeline(); MapString, Object resultMap new LinkedHashMap(); for (int i 0; i commands.size(); i) { resultMap.put(commands.get(i).getKey(), results.get(i)); } return resultMap; } catch (RedisException e) { log.error(Pipeline 混合操作异常, e); throw new PipelineExecutionException(Pipeline 混合操作失败, e); } finally { if (connection ! null) { connection.close(); } } } }2.2 Pipeline 的使用限制Pipeline 虽好但有四个需要关注的限制其一无原子性保证。Pipeline 中的命令在服务端是独立执行的其他客户端的命令可能穿插其间。如果你的业务要求要么全成功、要么全不执行Pipeline 做不到。其二单次 Pipeline 不宜过大。一次性发送数万条命令会导致客户端内存暴涨和服务端输出缓冲区溢出。建议单次控制在 500~1000 条命令以内超过则分批。其三错误不中断。Pipeline 中某一条命令失败其余命令仍会继续执行。必须在客户端逐一检查每个响应不能假设全部成功。其四集群模式下的跨槽限制。Redis Cluster 要求每条命令的所有 key 落在同一个 slot。跨 slot 的 Pipeline 需要客户端做分组——将命令按 slot 分组后每组独立发送。三、Lua 脚本的原子性保证与服务端执行3.1 Lua 脚本的核心优势Lua 脚本真正区别于 Pipeline 的特性是脚本内的所有命令在 Redis 服务端原子的、串行的执行。执行期间Redis 不会处理任何其他客户端的命令。这保证了操作的隔离性。/** * Redis Lua 脚本执行服务 * 封装脚本加载、执行、回滚逻辑 */ Service Slf4j public class RedisLuaService { private final StringRedisTemplate stringRedisTemplate; private final DefaultRedisScriptLong inventoryDeductScript; private final DefaultRedisScriptListObject batchCompareAndSetScript; public RedisLuaService(StringRedisTemplate stringRedisTemplate) { this.stringRedisTemplate stringRedisTemplate; // 库存扣减脚本原子校验 扣减 this.inventoryDeductScript new DefaultRedisScript(); this.inventoryDeductScript.setScriptText( local key KEYS[1] local deductAmount tonumber(ARGV[1]) local currentStock tonumber(redis.call(GET, key) or 0) if currentStock deductAmount then redis.call(DECRBY, key, deductAmount) return currentStock - deductAmount -- 返回扣减后的库存 else return -1 -- 库存不足 end ); this.inventoryDeductScript.setResultType(Long.class); // 批量 Compare-And-Set 脚本多 key 原子校验写入 this.batchCompareAndSetScript new DefaultRedisScript(); this.batchCompareAndSetScript.setScriptText( local results {} local count #KEYS / 2 for i 1, count do local key KEYS[i] local expectedValue KEYS[count i] local newValue ARGV[i] local ttl tonumber(ARGV[count i]) or 0 local currentValue redis.call(GET, key) if currentValue expectedValue then if ttl 0 then redis.call(SETEX, key, ttl, newValue) else redis.call(SET, key, newValue) end results[i] 1 -- 成功 else results[i] 0 -- CAS 失败 end end return results ); this.batchCompareAndSetScript.setResultType(List.class); } /** * 原子库存扣减 * * param productId 商品ID * param deductAmount 扣减数量 * return 扣减后库存-1 表示库存不足 */ public long deductInventory(String productId, long deductAmount) { if (deductAmount 0) { throw new IllegalArgumentException(扣减数量必须大于0); } try { String key inventory: productId; Long result stringRedisTemplate.execute( inventoryDeductScript, List.of(key), String.valueOf(deductAmount) ); if (result null) { log.error(库存扣减脚本返回null: productId{}, productId); throw new LuaExecutionException(脚本执行异常); } if (result -1) { log.warn(库存不足: productId{}, deductAmount{}, productId, deductAmount); throw new InsufficientInventoryException(productId, deductAmount); } log.info(库存扣减成功: productId{}, deductAmount{}, remaining{}, productId, deductAmount, result); return result; } catch (InsufficientInventoryException e) { throw e; } catch (Exception e) { log.error(库存扣减脚本执行异常: productId{}, productId, e); throw new LuaExecutionException(库存扣减失败, e); } } /** * 批量 CAS 更新 * 适用于分布式锁续期、配置版本校验等场景 */ public ListInteger batchCompareAndSet(MapString, CasEntry entries) { if (entries null || entries.isEmpty()) { return Collections.emptyList(); } ListString keys new ArrayList(); ListString expectedValues new ArrayList(); ListString newValues new ArrayList(); ListString ttls new ArrayList(); for (Map.EntryString, CasEntry entry : entries.entrySet()) { keys.add(entry.getKey()); expectedValues.add(entry.getValue().getExpectedValue()); newValues.add(entry.getValue().getNewValue()); ttls.add(String.valueOf(entry.getValue().getTtlSeconds())); } // KEYS [key1, key2, ..., expected1, expected2, ...] ListString allKeys new ArrayList(keys); allKeys.addAll(expectedValues); // ARGV [new1, new2, ..., ttl1, ttl2, ...] ListString allArgs new ArrayList(newValues); allArgs.addAll(ttls); try { SuppressWarnings(unchecked) ListObject rawResults stringRedisTemplate.execute( batchCompareAndSetScript, allKeys, allArgs.toArray(new String[0]) ); if (rawResults null) { throw new LuaExecutionException(CAS脚本返回null); } return rawResults.stream() .map(r - ((Number) r).intValue()) .collect(Collectors.toList()); } catch (Exception e) { log.error(批量CAS脚本执行异常, e); throw new LuaExecutionException(批量CAS失败, e); } } }3.2 Lua 脚本的注意事项一是不要执行耗时操作。Lua 脚本执行期间 Redis 是阻塞的复杂计算应该放在客户端完成。二是脚本尽可能幂等。如果脚本执行到一半客户端断连Redis 不会回滚已执行的命令。设计脚本时要考虑脚本被部分执行后重试的场景。三是集群模式下所有 key 必须在同一 slot。使用hash tag如{user:1001}:profile和{user:1001}:orders将关联 key 映射到同一 slot。四、Pipeline vs Lua 脚本的性能对比测试以下是在本地 Redis 6.2 上执行的对比测试JMH 基准测试操作INCR 1000 个 key/** * Pipeline vs Lua 性能基准测试 */ BenchmarkMode(Mode.AverageTime) OutputTimeUnit(TimeUnit.MILLISECONDS) State(Scope.Benchmark) Warmup(iterations 3, time 1) Measurement(iterations 5, time 2) Fork(1) public class PipelineVsLuaBenchmark { Param({100, 500, 1000, 5000}) private int batchSize; private StringRedisTemplate stringRedisTemplate; private DefaultRedisScriptListLong batchIncrScript; private ListString keys; Setup public void setup() { // 初始化 Redis 连接... keys new ArrayList(); for (int i 0; i batchSize; i) { keys.add(bench:key: i); } // Lua 批量 INCR 脚本 batchIncrScript new DefaultRedisScript( local results {} for i, key in ipairs(KEYS) do results[i] redis.call(INCR, key) end return results , List.class); } Benchmark public ListObject pipelineIncr() { // Pipeline 方式执行 RedisConnection conn RedisConnectionUtils.getConnection(connectionFactory); conn.openPipeline(); for (String key : keys) { conn.incr(key.getBytes(StandardCharsets.UTF_8)); } return conn.closePipeline(); } Benchmark public ListLong luaBatchIncr() { // Lua 脚本方式执行 return stringRedisTemplate.execute(batchIncrScript, keys); } }典型测试结果batchSize1000本地环境指标PipelineLua 脚本差异平均耗时1.18ms0.82msLua 快 30%P99 耗时2.41ms1.53msLua 快 37%服务端 CPU12%8%Lua 更低Lua 脚本的优势在 batchSize 增大时更明显——5000 条命令时 Lua 比 Pipeline 快约 45%。原因在于 Pipeline 虽然只发一次请求但服务端仍要逐条解析和处理命令而 Lua 脚本是一次性解析、内部循环执行减少了解析开销。五、总结Pipeline 和 Lua 脚本的选择有一个简单的决策框架只需要减少 RTT、不需要原子性→ 用 Pipeline。典型场景批量写入缓存、预热数据。需要原子性保证→ 用 Lua 脚本。典型场景库存扣减、CAS 更新、分布式限流。需要在服务端做条件判断如库存够就扣不够就返回错误→ 必须用 Lua 脚本。集群模式、key 分散在不同 slot→ Pipeline 分组后使用或使用 hash tag 规划 key 分布。两种方式可以组合使用在大批量数据加载场景中先用 Lua 脚本对每个 slot 内的数据做原子操作再通过 Pipeline 将各 slot 的操作并行发送兼顾原子性和吞吐。关于集群模式下的 hash tag 技巧一定要在项目初期就规划好 key 的命名规范。等到数据量上来再改 key 名称迁移成本远高于提前设计。一个实用的建议将需要关联操作的 key 统一使用{业务域:业务ID}:属性的格式如{order:123456}:status、{order:123456}:items这样哈希计算只依赖花括号内的部分确保它们落在同一 slot。