5Redis高级特性（慢查询Pipeline事务Lua）

Posted 2022-12-06 ＊King＊

一、慢查询

慢查询日志就是系统在命令执行前后计算每条命令的执行时间，当超过预设阀值,就将这条命令的相关信息（例如:发生时间，耗时，命令的详细信息）记录下来，Redis 也提供了类似的功能。

Redis客户端执行一条命令分为4个部分：

1）发送命令 2）命令排队 3）命令执行 4）返回结果

慢查询只统计步骤3的时间，所以没有慢查询并不代表客户端没有超时问题。

1、慢查询配置

Redis提供了 slowlog-log-slower-than 和 slowlog-max-len 配置

slowlog-log-slower-than预设阀值，它的单位是微秒（1秒=1000毫秒=1000 000微秒），默认值是10000，假如执行一条很慢的命令（例keys *），如果它的执行时间超过了10000微秒，也就是10毫秒，那么它将被记录在慢查询日志中。

slowlog-max-len 用来设置慢查询日志最多存储多少条，并没有说明存放在哪。实际上 Redis 使用了一个列表来存储慢查询日志，slowlog-max-len 就是列表的最 大长度。当慢查询日志列表被填满后，新的慢查询命令则会继续入队，队列中的第一条数据机会出列。

设置慢查询配置有两种方式：

1、执行以下命令

config set slowlog-log-slower-than 10000  //10毫秒

使用config set完后，苦想将配置持久化保存到Redis.conf，要执行

config rewrite

2、修改配置文件

Redis.conf修改：找到slowlog-log-slower-than 10000，修改保存即可。slowlog-log-slower-than =0记录所有命令-1命令都不记录。

2、慢查询操作命令

slowlog get  //获取队列里慢查询的命令
slowlog len  //获取慢查询列表当前的长度
slowlog reset //慢查询列表清理（重置），执行后再查slowlog len此时返回0清空
slow-max-len配置建议：线上可设置1000以上
slowlog-log-slower-than配置建议：默认为10毫秒，根据redis并发量来调整，对于高并发比建议为1毫秒

慢查询是先进先出的队列，访问日志记录出列丢失，需定期执行slowlog get，将结果存储到其它设备中（如mysql）

二、Pipeline

Pipeline（流水线）机制：它能将一组Redis命令进行组装，通过一次RTT（RTT:往返时间，也就是数据在网络上传输的时间）传输给Redis，再将这组Redis命令的执行结果按顺序返回给客户端，没有使用Pipeline执行了n条命令，整个过程需要n次RTT。使用了Pipeline执行了n次命令，整个过程需要1次RTT。

代码样例：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;

import java.util.List;

@Component
public class RedisPipeline 

    @Autowired
    private JedisPool jedisPool;

    public List<Object> plGet(List<String> keys) 
        Jedis jedis = null;
        try 
            jedis = jedisPool.getResource();
            Pipeline pipelined = jedis.pipelined();

            for(String key:keys)
                pipelined.get(key);
            
            return pipelined.syncAndReturnAll();
         catch (Exception e) 
            throw new RuntimeException("执行Pipeline获取失败！",e);
         finally 
            jedis.close();
        
    

    public void plSet(List<String> keys,List<String> values) 
        if(keys.size()!=values.size()) 
            throw new RuntimeException("key和value个数不匹配！");
        
        Jedis jedis = null;
        try 
            jedis = jedisPool.getResource();
            Pipeline pipelined = jedis.pipelined();
            for(int i=0;i<keys.size();i++)
                pipelined.set(keys.get(i),values.get(i));
            
            pipelined.sync();
         catch (Exception e) 
            throw new RuntimeException("执行Pipeline设值失败！",e);
         finally 
            jedis.close();
        
    

测试类

import cn.enjoyedu.redis.adv.RedisPipeline;
import cn.enjoyedu.redis.redisbase.basetypes.RedisString;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.ArrayList;
import java.util.List;

@SpringBootTest
public class TestRedisPipeline 

    @Autowired
    private RedisPipeline redisPipeline;
    @Autowired
    private RedisString redisString;

    private static final int TEST_COUNT = 1000;

    @Test
    public void testPipeline() 
        long setStart = System.currentTimeMillis();
        for (int i = 0; i < TEST_COUNT; i++) 
            redisString.set("testStringM:key_" + i, String.valueOf(i));
        
        long setEnd = System.currentTimeMillis();
        System.out.println("非pipeline操作"+TEST_COUNT+"次字符串数据类型set写入，耗时：" + (setEnd - setStart) + "毫秒");

        List<String> keys = new ArrayList<>(TEST_COUNT);
        List<String> values= new ArrayList<>(TEST_COUNT);
        for (int i = 0; i < keys.size(); i++) 
            keys.add("testpipelineM:key_"+i);
            values.add(String.valueOf(i));
        
        long pipelineStart = System.currentTimeMillis();
        redisPipeline.plSet(keys,values);
        long pipelineEnd = System.currentTimeMillis();
        System.out.println("pipeline操作"+TEST_COUNT+"次字符串数据类型set写入，耗时：" + (pipelineEnd - pipelineStart) + "毫秒");
    

测试结果：

三、事务

Redis提供了简单的事务功能，将一组需要一起执行的命令放到multi和exec两个命令之间，multi命令代表事务开始，exec命令代表事务结束，如果要停止事务的执行，可以使用discard命令代替exec命令即可。

可以看到sadd命令此时的返回结果是QUEUED，代表命令并没有真正执行，而是暂时保存在Redis中的一个缓存队列，所以 discard 也只是丢弃这个缓存队列中的未执行命令，并不会回滚已经操作过的数据，这一点要和关系型数据库的Rollback 操作区分开。只有当exec执行后，返回两个结果对应的sadd命令。

1、如果命令错误，比如set写成了sett，属于语法错误，会造成整个事务无法执行。

2、如果运行时错误，比如用户B在添加数据时，误把sadd命令写成了zadd命令，这种就是运行时命令，此时Redis并不支持回滚功能。

有时需要在事务之间，确保事务中的key没有被其他客户端修改过才执行事务，否则不执行（类似乐观锁）。Redis提供了watch命令来解决这类问题

客户端1：

客户端2：

再回到客户端1

可以看到客户端1在执行multi之前执行了watch命令，客户端2在客户端1执行exec之间修改了key值，造成了客户端1事务没有执行exec结果为nil

Redis客户端中的事务使用代码：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;

import java.util.List;

@Component
public class RedisTransaction 

    public final static String RS_TRANS_NS = "rts:";

    @Autowired
    private JedisPool jedisPool;

    public List<Object> transaction(String... watchKeys)
        Jedis jedis = null;
        try 
            jedis = jedisPool.getResource();
            if(watchKeys.length>0)
                /*使用watch功能*/
                String watchResult = jedis.watch(watchKeys);
                if(!"OK".equals(watchResult)) 
                    throw new RuntimeException("执行watch失败:"+watchResult);
                
            
            Transaction multi = jedis.multi();
            multi.set(RS_TRANS_NS+"testa1","a1");
            multi.set(RS_TRANS_NS+"testa2","a2");
            multi.set(RS_TRANS_NS+"testa3","a3");
            List<Object> execResult = multi.exec();
            if(execResult==null)
                throw new RuntimeException("事务无法执行，监视的key被修改:"+watchKeys);
            
            System.out.println(execResult);
            return execResult;
         catch (Exception e) 
            throw new RuntimeException("执行Redis事务失败！",e);
         finally 
            if(watchKeys.length>0)
                jedis.unwatch();/*前面如果watch了，这里就要unwatch*/
            
            jedis.close();
        
    

测试：

@SpringBootTest
public class TestRedisTransaction 

    @Autowired
    private RedisTransaction redisTransaction;

    @Test
    public void testTransaction() 
        redisTransaction.transaction();
    

测试结果：

四、Lua

LUA脚本语言是C开发的，类似存储过程。

使用LUA脚本的好处：

• 减少网络开销，在Lua脚本中可以把多个命令放在同一个脚本中运行
• 原子操作，Redis会将整个脚本作为一个整体执行，中间不会被其他命令插入
• 复用性，客户端发送的脚本会存储在Redis中，其他客户端可以复用这个脚本来完成同样的逻辑

1、安装Lua

Lua在Linux中的安装：

1、下载lua的tar.gz的源码包

wget http://www.lua.org/ftp/lua-5.3.6.tar.gz

2、解压

tar -zxvf lua-5.3.6.tar.gz 

3、进入解压目录

cd lua-5.3.6
make linux
make install   //需要在root用户下运行

如报错，找不到readline/readline.h，可以root用户下通过yum命令安装

yum -y install libtermcap-devel ncurses-devel libevent-devel readline-devel

安装完成再

make linux
make install

最后，直接进入lua命令即可进入lua的控制台：

2、Lua基本语法

• 单行注释： –
• 多行注释： --[[ --]]
• Lua关键字：

and break do else elseif end false for function if in local nil not or repeat return then true until while

Lua数据类型：lua是动态类型语言，变量不要类型定义，只需要为变量赋值，值可以存储在变量中，作为参数传递或结果返回。

Lua中有8个基本类型：

• nil 表示一个无效值（在条件表达式中相当于false）

• boolean 包含两个值 : false和true

• number 表示双精度类型的实浮点数

• string 字符串由一对双引号或单引号来表示，也可以[[与]]表示

• function 由c或lua编写的函数

• userdata 表示任意存储在变量中的c数据结构

• thread 表示执行的独立线数，用于执行协同程序

• table lua中的表，可以做为数组，也可以作为hash
  
  

Lua中的函数

在lua中，函数以function开头，以end结尾，funcName是函数名，中间部分是函数体

function funcName () 
	--[[
	函数内容 
	--]] 
end

定义一个字符串连接函数：

Lua变量

Lua变量有全局变量和局部变量

lua中的变量全是全局变量，除非用local显式声明为局部变量的。局部变量的作用域为从声明位置开始到所在语句块结束。

变量的默认值均为nil。

Lua中的控制语句

循环控制：

Lua支持while循环、for循环，repeat…until循环和循环嵌套，同时lua提供了break语句和goto语句

数值for循环

for var=exp1,exp2,exp3 do
  <执行体>
end

var从exp1变化到exp2,每次变化以exp3为步长递寺var，并执行一次“执行体”。exp3是可选的，如果不指定，默认为1。

泛型for循环

例如：打印数组a的所有值

i 是数组索引值，v是对应索引的数组元素值。ipairs是lua提供的一个迭代器函数，用来迭代数组。

while 循环 
while(condition) 
do 
	statements 
end

if条件控制

Lua支持if语句、if…else语句和if嵌套语句

if语句语法：

if(布尔表达式) 
then
	--[ 在布尔表达式为 true 时执行的语句 --] 
end

if…else语句语法：

if(布尔表达式) 
then
	--[ 布尔表达式为 true 时执行该语句块 --] 
else
	--[ 布尔表达式为 false 时执行该语句块 --]
end

Lua运算符

算术运算符

+ 加法

- 减法

* 乘法

/ 除法

% 取余

^ 乘幂

- 负号

关系运算符

== 等于

~= 不等于

> 大于

< 小于

>= 大于等于

<= 小于等于

逻辑运算符

and 逻辑与操作符

or 逻辑或操作符

not 逻辑非操作符

3、Redis中的Lua

eval命令

EVAL script numkeys key [key ...] arg [arg ...]

命令说明：

• （1）script 参数：是一段 Lua 脚本程序，它会被运行在 Redis 服务器上下文中，这段脚本不必（也不应该）定义为一个 Lua 函数。
• （2）numkeys 参数：用于指定键名参数的个数。
• （3）key [key …] 参数： 从 EVAL 的第三个参数开始算起，使用了 numkeys 个键（key），表示在脚本中所用到的那些 Redis 键（key），这些键名参数可以在 Lua 中通过全局变量 KEYS 数组，用 1 为基址的形式访问（KEYS[1],KEYS[2]···）。
• （4）arg [arg …]参数：可以在 Lua 中通过全局变量 ARGV 数组访问，访问的形式和 KEYS 变量类似（ARGV[1],ARGV[2]···）。
  

Lua脚本中调用Redis命令

eval "return redis.call('mset',KEYS[1],ARGV[1],KEYS[2],ARGV[2])" 2 key1 key2 first second
eval "return redis.call('set',KEYS[1],ARGV[1])" 1 key1 newfirst

evalsha命令

• script flush ：清除所有脚本缓存。
• script exists ：根据给定的脚本校验，检查指定的脚本是否存在于脚本缓存。
• script load ：将一个脚本装入脚本缓存，返回 SHA1 摘要，但并不立即运行它。
• script kill ：杀死当前正在运行的脚本。
  

3、Java对Lua的支持

在java生态中，对Lua的支持是LuaJ，是一个java的Lua解释器，基于Lua5.2.x版本

Java客户端使用Lua脚本

Maven依赖

<dependency> 
	<groupId>org.luaj</groupId> 
	<artifactId>luaj-jse</artifactId> 					<version>3.0.1</version> 
</dependency>

案例代码

/*基于redis的一个限流功能*/
@Component
public class RedisLua 

    public final static String RS_LUA_NS = "rlilf:";
    /*第一次使用incr对KEY（某个IP作为KEY）加一，如果是第一次访问，
    使用expire设置一个超时时间，这个超时时间作为Value第一个参数传入，
    如果现在递增的数目大于输入的第二个Value参数，返回失败标记，否则成功。
    redis的超时时间到了，这个Key消失，又可以访问
        local num = redis.call('incr', KEYS[1])
        if tonumber(num) == 1 then
            redis.call('expire', KEYS[1], ARGV[1])
            return 1
        elseif tonumber(num) > tonumber(ARGV[2]) then
            return 0
        else
            return 1
        end
    * */
    public final static String LUA_SCRIPTS =
            "local num = redis.call('incr', KEYS[1])\\n" +
            "if tonumber(num) == 1 then\\n" +
            "\\tredis.call('expire', KEYS[1], ARGV[1])\\n" +
            "\\treturn 1\\n" +
            "elseif tonumber(num) > tonumber(ARGV[2]) then\\n" +
            "\\treturn 0\\n" +
            "else \\n" +
            "\\treturn 1\\n" +
            "end";

    @Autowired
    private JedisPool jedisPool;

    public String loadScripts()
        Jedis jedis = null;
        try 
            jedis = jedisPool.getResource();
            String sha =jedis.scriptLoad(LUA_SCRIPTS);
            return sha;
         catch (Exception e) 
            throw new RuntimeException("加载脚本失败！",e);
         finally 
            jedis.close();
        
    

    public String ipLimitFlow(String ip)
        Jedis jedis = null;
        try 
            jedis = jedisPool.getResource();
            String result = jedis.evalsha("9ac7623ae2435baf9ebf3ef4d21cde13de60e85c",
                    Arrays.asList(RS_LUA_NS+ip),Arrays.asList("60","2")).toString();
            return result;
         catch (Exception e) 
            throw new RuntimeException("执行脚本失败！",e);
         finally 
            jedis.close();
        
    

测试

@SpringBootTest
public class TestRedisLua 

    @Autowired
    private RedisLua redisLua;

    @Test
    public void testLoad() 
        System.out.println(redisLua.loadScripts());
    

    @Test
    public void tesIpLimitFlow() 
        System.out.println(redisLua.ipLimitFlow("localhost"));
    

测试结果