Redis 性能优化实战：Pipeline 和 multiGet

Redis 性能优化实战：Pipeline 和 multiGet

循环里调redisTemplate.opsForValue().get(key)——每次都是独立的一次网络往返。100 个 key 就是 100 次网络 IO，99 次纯浪费在 TCP 握手和等待上。Redis 自己一毫秒就查完了，但你花了 100 毫秒在路上。Pipeline 把一堆命令打包发过去一口气执行，multiGet 把多个 GET 合并成一条命令——两种手段，干的同一件事：砍掉多余的往返。

没有 Pipeline 的时候，循环 GET 有多慢

// 经典反例：循环 GET——100 个 key = 100 次 Redis 请求for(Stringkey:keys){Stringvalue=redisTemplate.opsForValue().get(key);// 每次都是独立的网络往返// ...}

Redis 单线程，执行一次 GET 不到 1ms。但网络往返每次 1ms，100 个 key 累计 100ms。耗时不在 Redis 身上——全花在来回的网络上。

方案一：multiGet——合并成一条命令，一次往返查完

multiGet把多个 GET 合并成一条 Redis 命令MGET，一次网络往返拿回所有结果。

从 100 次往返变成 1 次。耗时从 100ms 降到 ~2ms。

代码

// 构建 Redis Key 列表List<String>redisKeys=userIds.stream().map(RedisKeyConstants::buildUserInfoKey).toList();// 一次 multiGet，查完所有 keyList<Object>redisValues=redisTemplate.opsForValue().multiGet(redisKeys);// 过滤掉不存在的 key（值为 null）redisValues=redisValues.stream().filter(Objects::nonNull).toList();

实战常用：批量查完，缺的再走数据库

不是所有 key 都命中了缓存，对没命中的部分降级查 DB：

// 先从 Redis 批量查List<Object>redisValues=redisTemplate.opsForValue().multiGet(redisKeys);// 把命中的转成结果集List<UserVO>result=redisValues.stream().filter(Objects::nonNull).map(v->JsonUtils.parseObject(v.toString(),UserVO.class)).toList();// 全命中了——直接返回if(result.size()==userIds.size()){returnresult;}// 没全命中——找出缺失的 userId，查 DBList<Long>missedIds=userIds.stream().filter(id->result.stream().noneMatch(u->u.getId().equals(id))).toList();List<UserVO>dbResults=userMapper.selectByIds(missedIds);result.addAll(dbResults);// 异步把 DB 查到的回种到 RedisthreadPool.submit(()->{for(UserVOuser:dbResults){Stringkey=buildUserInfoKey(user.getId());longexpire=24*3600+ThreadLocalRandom.current().nextInt(24*3600);redisTemplate.opsForValue().set(key,JsonUtils.toJsonString(user),expire,TimeUnit.SECONDS);}});returnresult;

关键点：过期时间加了随机值。如果不加随机值，100 个 key 都在同一秒过期，下一秒全打在 DB 上——这就是缓存雪崩。

方案二：Pipeline——多条不同类型的命令打包一起执行

multiGet只能干一件事：批量 GET。如果你需要混合操作——比如批量 SET 100 个不同 value，或者 SET + EXPIRE 一起，multiGet 就管不了了。

Pipeline 不限制命令类型：你可以把任意数量的 GET、SET、DEL、EXPIRE……打包成一个批次，一次网络往返发给 Redis。

代码

// 异步线程将用户信息批量同步到 Redis 中threadPoolTaskExecutor.submit(()->{Map<Long,UserVO>map=dbResults.stream().collect(Collectors.toMap(UserVO::getId,u->u));// 执行 Pipeline——一批 SET 命令打包发送redisTemplate.executePipelined(newSessionCallback<>(){@OverridepublicObjectexecute(RedisOperationsoperations){for(UserDOuser:userList){Stringkey=RedisKeyConstants.buildUserInfoKey(user.getId());Stringvalue=JsonUtils.toJsonString(map.get(user.getId()));longexpire=24*3600+RandomUtil.randomInt(24*3600);// 这里的 SET 不会立刻发到 Redis——攒在 Pipeline 里operations.opsForValue().set(key,value,expire,TimeUnit.SECONDS);}returnnull;// Pipeline 里的返回值不关心}});});

关键细节

Pipeline 里的命令不会立即发送。executePipelined里所有的operations.xxx()调用只是把命令加入缓冲区，execute方法结束后，这批命令一次性发给 Redis。

Pipeline 不保证原子性。跟 Lua 脚本不一样——Pipeline 只是"打包发送"，Redis 接收后逐条执行，中间可能被其他客户端插一条命令。需要原子性还是得用 Lua。

返回值为 null。Pipeline 模式下每条命令的结果可以不收集（return null），因为我们只关心"写进去没有"，不关心返回值。如果想知道每一条的执行结果，可以收集但不常用。

三种方式对比

日常使用：查多用 multiGet，写多用 Pipeline。