redis之批量操作 pipeline重写

╰+哭是因爲堅強的太久メ 2023-06-12 06:25 1阅读 0赞

**目录**

起因

redis批量操作

批量操作产生的问题

解决办法

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

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

--------------------

## **起因 ** ##

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70][]

该操作做了大量的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);
        }

可以成功返回，

拦截请求如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 1][]

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

## 批量操作产生的问题 ##

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 2][]

@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（管道）][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方式后，速度确实上去了不少，但是根据监控链路的查看，发现有少量的调用链路出现了如下图问题：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 4][]

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

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

[https://github.com/lettuce-io/lettuce-core/issues/971][https_github.com_lettuce-io_lettuce-core_issues_971]

[https://github.com/lettuce-io/lettuce-core/issues/1027][https_github.com_lettuce-io_lettuce-core_issues_1027]

[https://github.com/reactor/reactor-netty/issues/564][https_github.com_reactor_reactor-netty_issues_564]

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

我太南了

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

[https://blog.csdn.net/dongying1751/article/details/103812175][https_blog.csdn.net_dongying1751_article_details_103812175]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70]: /images/20230531/f5c8fa4a58e7469d95042e042deddf0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 1]: /images/20230531/cac70f3e06f6479b913be3e79a83bba0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 2]: /images/20230531/2916c69fc7824a528fc79eeaac0e9c70.png
[Redis_Pipeline]: https://blog.csdn.net/u011489043/article/details/78769428
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 3]: /images/20230531/f21e06284ebb4e50b62b28fced71281b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5aW5nMTc1MQ_size_16_color_FFFFFF_t_70 4]: /images/20230531/f45d1332c57e42bf875d54a70eb20ab9.png
[https_github.com_lettuce-io_lettuce-core_issues_971]: https://github.com/lettuce-io/lettuce-core/issues/971
[https_github.com_lettuce-io_lettuce-core_issues_1027]: https://github.com/lettuce-io/lettuce-core/issues/1027
[https_github.com_reactor_reactor-netty_issues_564]: https://github.com/reactor/reactor-netty/issues/564
[https_blog.csdn.net_dongying1751_article_details_103812175]: https://blog.csdn.net/dongying1751/article/details/103812175