Springboot使用lua脚本存储redis数据及问题

墨蓝 2022-12-21 15:00 110阅读 0赞

springboot 使用rua脚本执行redis的命令记录

## 1.问题一：在lua脚本中，if条件比较？ ##

-- 如果传入的最大时间大于redis的最大时间 更新 lua脚本数组下标从1开始
    local redisMaxTime = redis.call("get", KEYS[1])
    -- 如果不为空，再去比较时间大小，redis存值为90，传过来的参数为690，执行完代码，redis的值未更新
    if ARGV[1] > redisMaxTime then
    	redis.call("set", KEYS[1], ARGV[1])
    end

## 现象：690 > 90 按照条件会更新redis的存值，结果却一直未更新！！！ ##

过程：在网上查阅资料lua脚本的if判断没有写错，将 > 改成 ~= 不等于就没有问题，但是需求是更新对应key的最大值。  
一路查看资料未果，在spring开发群也咨询了一些大佬，提示存储值应为int，这样好比较。  
redisTemplate.opsForValue().set(key, 10, 3600, TimeUnit.SECONDS);  
设置了key对应的值为10,3600s过期。

但是if条件还是不对，redis的值还是未更新最大值，在网上看了很久，几近崩溃了，周末也一直在想，为啥690 > 90 不成立呢0.0  
最终问了下唐司机，卧槽他让我试下将里面的值tonumber(),特么竟然对了，卧槽，来看下原因：特么从redis取出来的值得类型是  
object，卧槽！！我存的是int类型，我以为取出来也是int，大意了！！哎，被这个坑了好久啊，泪崩。看来很多细节还是没有处理好，  
以为是存的int取出来也是int，疏忽了~~  
解决方案：if tonumber(ARGV\[1\]) > tonumber(redisMaxTime) then

## 2.问题二：redis集群使用以下方式失败！提示集群不支持…（我自己的本机是单机模式，测试通过，但是在集群环境就用不了…） ##

public void updateMaxMinTime(String maxTimeKey, String minTimeKey, Integer consumeTime) { 
            // 执行 lua 脚本
            DefaultRedisScript<Object> redisScript = new DefaultRedisScript<>();
            // 指定 lua 脚本
            // 先获取指定key的值，然后和传入的arg比较是否相等，相等值删除key，否则直接返回0。
            redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("script/updateMaxMinTime.lua")));
            // 指定返回类型
            redisScript.setResultType(Object.class);
            List<String> keyList = new ArrayList<>();
            keyList.add(maxTimeKey);
            keyList.add(minTimeKey);
            // 参数一：redisScript，参数二：key列表，参数三：arg（可多个）
            redisTemplate.execute(redisScript, keyList, consumeTime);
            log.info("执行updateMaxTime.lua脚本结束.");
        }

解决方案：查阅资料，使用以下方式。

public void updateMaxMinTime1(String maxTimeKey, String minTimeKey, Integer consumeTime) { 
            List<String> keys = new ArrayList<>();
            keys.add(maxTimeKey);
            keys.add(minTimeKey);
            List<String> args = new ArrayList<>();
            args.add(consumeTime.toString());
            //spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本异常，此处拿到原redis的connection执行脚本
            Object result = redisTemplate.execute(new RedisCallback<Object>() { 
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException { 
                    Object nativeConnection = connection.getNativeConnection();
                    // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                    // 集群模式
                    if (nativeConnection instanceof JedisCluster) { 
                        return ((JedisCluster) nativeConnection).eval(LUA, keys, args);
                    }
                    // 单机模式
                    else if (nativeConnection instanceof Jedis) { 
                        return ((Jedis) nativeConnection).eval(LUA, keys, args);
                    }
                    return null;
                }
            });
            log.info("执行updateMaxTime.lua脚本结束.{}", result);
        }

**但是问题又来了，redis.clients.jedis.exceptions.JedisClusterException: 
    			No way to dispatch this command to Redis Cluster because keys have different slots.

我网上看了下 有人说jedisCluster不支持mget/mset等跨槽位的操作  
需要更改redis的驱动为lettuce ~~\*\*

<dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
            </dependency>
    
    
    		
            <dependency>
                <groupId>io.lettuce</groupId>
                <artifactId>lettuce-core</artifactId>
                <version>5.1.8.RELEASE</version>
            </dependency>

将使用lettuce替换jedis的客户端…但是问题又来了，CROSSSLOT Keys in request don’t hash to the same slot

会报此错误：CROSSSLOT Keys in request don’t hash to the same slot–keys不能落在同一个节点  
上面的代码传入了两个key，key1、key2，但算slot的时候发现并未落在同一个slot里面。  
针对提示，那就让keys落在同一个slot里面就行啦  
使用hash tag(hash tag是用于hash的部分字符串开始和结束的标记)，即如果key的结构为\{XXX\}key，  
则只会对\{\}里面的XXX进行分区，就会落到同一个slot里面啦。

**修改方案：key添加时，前面添加\{xxx\}  
List keys = new ArrayList<>();  
keys.add("\{STATISTICS\}" + maxTimeKey);  
keys.add("\{STATISTICS\}" + minTimeKey);  
最终执行：redis存储成功！！！**

## 3.问题三：如果使用lettuce链接redis，会出现：lettuce在docker中会出问题，没有连接保活，240s后面就用不了redis了 ##

what？？？同事说会有这个风险，让尝试：集群模式Redis支持pipeline  
时间原因就没有去尝试pipline了。  
将lettuce依赖删除，还是使用jedis，还是使用eval的方式执行lua脚本

public void updateMaxMinTime(String maxTimeKey, String minTimeKey, String successKey, String failKey, Integer consumeTime) { 
            try { 
                // 设置key列表
                List<String> keys = new ArrayList<>();
                keys.add(applicationName + SEPARATOR + STATISTICS + maxTimeKey);
                keys.add(applicationName + SEPARATOR + STATISTICS + minTimeKey);
                keys.add(applicationName + SEPARATOR + STATISTICS + successKey);
                keys.add(applicationName + SEPARATOR + STATISTICS + failKey);
                // 设置参数列表
                List<String> args = new ArrayList<>();
                args.add(consumeTime.toString());
                redisTemplate.execute(new RedisCallback<Object>() { 
                    @Override
                    public Object doInRedis(RedisConnection connection) throws DataAccessException { 
                        Object nativeConnection = connection.getNativeConnection();
                        // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                        // 集群模式
                        if (nativeConnection instanceof JedisCluster) { 
                            return ((JedisCluster) nativeConnection).eval(lua_script, keys, args);
                        }
                        // 单机模式
                        else if (nativeConnection instanceof Jedis) { 
                            return ((Jedis) nativeConnection).eval(lua_script, keys, args);
                        }
                        return null;
                    }
                });
                log.info("执行updateMaxTime.lua脚本结束。");
            } catch (Exception e) { 
                log.error("执行updateMaxTime.lua脚本失败。");
            }
        }

脚本：

-- 如果传入的最大时间大于redis的最大时间 更新 lua脚本数组下标从1开始
    	local redisMaxTime = redis.call("get", KEYS[1])
    	-- 如果不为空，再去比较时间大小
    	if redisMaxTime then
    		if tonumber(ARGV[1]) > tonumber(redisMaxTime)
    		then
    			redis.call("set", KEYS[1], ARGV[1])
    		end
    	-- 如果为空，则说明是第一次存储，直接存入redis
    	else
    		redis.call("set", KEYS[1], ARGV[1])
    	end
    
    	-- 如果传入的最小时间小于redis的最小时间 更新
    	local redisMinTime = redis.call("get", KEYS[2])
    	if redisMinTime then
    		if tonumber(ARGV[1]) < tonumber(redisMinTime)
    		then
    			redis.call("set", KEYS[2], ARGV[1])
    		end
    	else
    		redis.call("set", KEYS[2], ARGV[1])
    	end
    
    	-- 成功次数key是否对应值，没有则设置为0
    	if not redis.call("get", KEYS[3]) then
    		redis.call("set", KEYS[3], 0)
    	end
    
    	-- 失败次数key是否对应值，没有则设置为0
    	if not redis.call("get", KEYS[4]) then
    	   redis.call("set", KEYS[4], 0)
    	end