redis之批量操作pipeline重写

目录

 

起因 

redis批量操作

批量操作产生的问题

解决办法

2019年11月5日更新（使用pipeline改写调用redis）

2019年12月11日更新（pipeline调用产生的问题）

 

 

---

 

起因 

最近在测试环境上发现了一个比较慢的链路调用，如下图所示：

该操作做了大量的hgetall，发现入参中的List参数高达2700个，导致这里操作redis的时候一直循环查询，

因为这个接口在设计的时候没有料想到有如此巨大的数据入口（所以没有限制入口参数列表长度，也没有优化循环操作），

导致该调用链直接耗时16s之久；经过调查，决定重写redis查询接口；

 

redis批量操作

查阅资料得知，redis有批量操作，org.springframework.data.redis.core.ValueOperations#multiGet

写自测例子如下：

@Autowired private StringRedisTemplate stringRedisTemplate; @GetMapping("/combine") @ApiOperation(value = "组合服务测试") public R<?> combine() { Map<String, String> map = IntStream.rangeClosed(1, 100).boxed() .collect(Collectors.toList()).stream() .collect(Collectors.toMap(k -> "test-app-" + k, v -> "" + v)); redisTemplate.opsForValue().multiSet(map); List<String> collect = IntStream.rangeClosed(1, 100).boxed().map(p -> "test-app-" + p).collect(Collectors.toList()); List<String> list = stringRedisTemplate.opsForValue().multiGet(collect); return R.success(list); }

可以成功返回，

拦截请求如下：

 

StringRedisTemplate 有multiget方法 1. 会把所有key所在的集群槽位分组 2. 每个槽位取一个redis连接 3. 没有开启pipeline, 从上个抓包图可以看出来, 每个请求都有回应

 

批量操作产生的问题

但是这里存在一个问题：只能操作String类型的数据，其他数据均会返回null

这点从官网可以看出（其实最后都是用的MGET命令）

 

@Override public List<V> multiGet(Collection<K> keys) { if (keys.isEmpty()) { return Collections.emptyList(); } byte[][] rawKeys = new byte[keys.size()][]; int counter = 0; for (K hashKey : keys) { rawKeys[counter++] = rawKey(hashKey); } List<byte[]> rawValues = execute(connection -> connection.mGet(rawKeys), true); return deserializeValues(rawValues); } /** * Get multiple {@code keys}. Values are returned in the order of the requested keys. * * @param keys must not be {@literal null}. * @return empty {@link List} if keys do not exist or when used in pipeline / transaction. * @see <a href="http://redis.io/commands/mget">Redis Documentation: MGET</a> */ @Nullable List<byte[]> mGet(byte[]... keys);

 

解决办法

如果要使用其他类型的批量数据操作，需要使用pipeline操作，所谓的pipeline操作其实就是

pipeline通过减少客户端与redis的通信次数来实现降低往返延时时间，而且Pipeline 实现的原理是队列，而队列的原理是时先进先出，这样就保证数据的顺序性。

主要思想就是尽可能的减少网络传输的时间损耗，以提高性能。

下面给出一篇文章具体详解redis的pipeline操作：分布式缓存Redis之Pipeline（管道）

具体实现接下来详解补充（待续）。。。

 

---

2019年11月5日更新（使用pipeline改写调用redis）

修改為pipeline方式獲取數據

List<String> stringList = new ArrayList<String>() { { add("xxx"); add("xxx"); } }; List<Object> List = stringStringRedisTemplate.executePipelined((RedisCallback<Object>) redisConnection -> { stringList.forEach(string->{ redisConnection.hVals(string.getBytes()); }); return null; });

这里是不需要openconnect和close的，看最后调用的redistemplate的方法org.springframework.data.redis.core.RedisTemplate#executePipelined(org.springframework.data.redis.core.RedisCallback<?>, org.springframework.data.redis.serializer.RedisSerializer<?>)

public List<Object> executePipelined(RedisCallback<?> action, @Nullable RedisSerializer<?> resultSerializer) { return (List)this.execute((connection) -> { connection.openPipeline(); boolean pipelinedClosed = false; List var7; try { Object result = action.doInRedis(connection); if (result != null) { throw new InvalidDataAccessApiUsageException("Callback cannot return a non-null value as it gets overwritten by the pipeline"); } List<Object> closePipeline = connection.closePipeline(); pipelinedClosed = true; var7 = this.deserializeMixedResults(closePipeline, resultSerializer, this.hashKeySerializer, this.hashValueSerializer); } finally { if (!pipelinedClosed) { connection.closePipeline(); } } return var7; }); }

這裏是獲取hash的value的方式，其他方式類似。

完结撒花...

 

2019年12月11日更新（pipeline调用产生的问题）

 

更改了pipeline方式后，速度确实上去了不少，但是根据监控链路的查看，发现有少量的调用链路出现了如下图问题：

 

目前暂未找到解决方法,未发现会产生什么负面影响

这里有个最相近的，并未给出解决方法：

https://github.com/lettuce-io/lettuce-core/issues/971

https://github.com/lettuce-io/lettuce-core/issues/1027

https://github.com/reactor/reactor-netty/issues/564

可能是因为生菜的问题吧。。。感觉更像是netty的问题，，，

我太南了

 

2020-1-7 更新 ........解决办法

https://blog.csdn.net/dongying1751/article/details/103812175