RedisTemplate常用API+事务+陷阱+序列化+pipeline+LUA

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› LUA›LUA教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

https://www.jianshu.com/p/7bf5dc61ca06/

https://blog.csdn.net/qq_34021712/article/details/79606551

https://www.jianshu.com/p/c9f5718e58f0

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

application.properties

简易测试：项目设置pool.max-active=10 ，客户端redis服务器都是配置很低的PC，无其他业务代码，查询内容在20个字符以内

用核心线程1000的线程池并发执行3000千次（Hash+String）查询，用时在1s左右。

用核心线程100的线程池并发执行3000千次（Hash+String）查询，用时在1100ms左右。

把设置改为pool.max-active=50，速度没有提升？可能是redis服务器端性能瓶颈？

# Redis数据库索引（默认为0）
spring.redis.database=0  
# Redis服务器地址
spring.redis.host=127.0.0.1
# Redis服务器连接端口
spring.redis.port=6379  
# Redis服务器连接密码（默认为空）
spring.redis.password=
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.pool.max-active=8  
# 连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.pool.max-wait=-1  
# 连接池中的最大空闲连接
spring.redis.pool.max-idle=8  
# 连接池中的最小空闲连接
spring.redis.pool.min-idle=0  
# 连接超时时间（毫秒）
spring.redis.timeout=0  

#哨兵的配置列表  配置哨兵就不用配置单独host
spring.redis.sentinel.master=mymaster
spring.redis.sentinel.nodes=39.107.119.256:26379
##哨兵集群
#spring.redis.sentinel.nodes=39.107.119.254:26379,39.107.119.254:26380

序列化Object

https://blog.csdn.net/u013958151/article/details/80603365

默认使用jdk的序列化 RedisTemplate<Object, Object>，不推荐使用JDK序列化，因为其在Redis服务器中是乱码不能解析，可读比较差。下图是JDK序列化后的KEY-VALUE在Redis服务器中的显示效果。

自定义序列化RedisTemplate<Object, Object>，redisTemplate的序列化一定要与redis中Value保存的序列化格式一致，才能在get时正确的反序列化，否则会抛出异常。

 /**
     * redisTemplate 序列化使用的jdkSerializeable, 存储二进制字节码, 所以自定义序列化类
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);

        // 使用Jackson2JsonRedisSerialize 替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);

        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        // 设置value的序列化规则和 key的序列化规则
        // value： 将Object转化为Json 保存在Redis中
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); //另外一种JSON格式new GenericJackson2JsonRedisSerializer()
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        //设置HASH序列化
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);

        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

下边所有测试都是基于上面redisTemplate的配置

以上序列化配置在Redis中的存储效果，String类型自动解析为""，自定义类型会记录全类名。

注意设置数字时要这样执行redisTemplate.opsForValue().set("1",1) ;

不要这样执行redisTemplate.opsForValue().set("1","1")；这样将设置为字符串

第二种设置的是字符串，不支持任何数字操作api

String-Json

SET-JSON

HASH

常用API

以下以RedisTemplate<String, String>为例

Key

redisTemplate.

keys(String pattern):Set<String>
hasKey(String k):Boolean
delete(String k)
expire(String k,long timeout,TimeUnit unit)
expireAt(String k,Date date)
getExpire(String k)
getExpire(String k,TimeUnit unit)
watch(String k)
watch(Collection<String> keys)
multi()
exec():List<Object>
discard()

String

redisTemplate.opsForValue()

set(String k, String v) :void
get(Object k) :String 
getAndSet(String k, String v) :String 
setIfAbsent(String k, String v) :Boolean //SETNX
increment(String k, long v) :Long //加v 返回运算后结果
increment(String k, double v) :Double //加v 返回运算后结果

List

redisTemplate.opsForList()

size(String k)
range(String k, long s, long e) :List<String> //返回list k的从s到e元素
leftPush(String k, String v) :Long  //还有对应的right版
leftPushIfPresent(String k, String v):Long
leftPop(String k) :String  //没有会阻塞
leftPop(String  key, long time, TimeUnit unit):String
set(String k,long index,String value)
remove(String k,long count,Object value):Long //移除count个，value相等的值，返回移除个数

Hash

redisTemplate.opsForHash()

put(String key, Object hashKey,Object value)
putIfAbsent(String key,Object hashKey,Object value):Boolean
delete(String key,Object... hashKeys):Long	
get(String key, Object hashKey):Object
increment(String key,Object hashKey,long delta):Long  //还有Double版 返回计算结果
size(String key):Long	
hasKey(key, hashKey):Boolean
keys(key):Set<Object>
values(key):List<Object>
entries(key):Map<Object,Object>

RedisTemplate事务

watch、unwatch、multi、exec、discard

可以实现CAS，非原子性事务批量执行，按顺序地串行化整体执行而不会被其它命令插入。multi命令使Redis将这些命令放入queue，而不是执行这些命令。当放入queue失败时（例如：语法错误）,EXEC命令可以检测到并且不执行。一旦调用EXEC命令成功，那么Redis就会一次性执行事务中所有命令，中间某条指令的失败不会导致前面已做指令的回滚，也不会造成后续的指令不做。相反，调用DISCARD命令将会清除事务队列，然后退出事务。EXEC命令的返回值是一个数组，其中的每个元素都分别是事务中的每个命令的返回值，返回值的顺序和命令的发出顺序是相同的。

//CAS 伪代码
//注意MULTI中所有命令 返回NULL，包括GET
WATCH mykey
val = GET mykey
val = val + 1
MULTI
SET mykey $val
EXEC

RedisTemplate事务陷阱

https://blog.csdn.net/qq_34021712/article/details/79606551（Spring 事务陷阱，永不关闭连接）

https://www.jianshu.com/p/c9f5718e58f0

陷阱1：不关闭连接问题

如果你的redisTemplate已经开启了事务支持 redisTemplate.setEnableTransactionSupport(true),

且在未标明@Transactional的方法内使用时，需要手动释放链接,否则链接一直不解绑关闭，造成无可用连接问题，或内存爆满

redisTemplate.setEnableTransactionSupport(true);

public Object get(String key){
	Object o = redisTemplate.opsForValue().get(key);
        //解绑 释放链接
	TransactionSynchronizationManager.unbindResource(redisTemplate.getConnectionFactory());
	return o;
}

陷阱2：事务将命令放到Queue中，因为命令并没有执行，所以无法拿到返回值,代码片段如下：

假设当前在multi事务中

String i =redisTemplate.opsForValue().get("6666").toString;
Long increment = redisTemplate.opsForValue().increment("6666", 1);

System.out.println(i);//null
System.out.println(increment);//null

是无法拿到返回值的，是因为redis在MULTI/EXEC代码块中，命令都会被delay，放入Queue中，而不会直接返回对应的值。即例子中的increment自增,自增命令执行完成,默认是会返回自增之后的值,但是却是返回了null.

解决思路

前提必须设置：redisTemplate.setEnableTransactionSupport(true)

方法1： @Transactional 来声明 Redis 事务的范围，中间发生异常全部不执行（因为Redis事务是放入队列，Exec时一起执行）

redisTemplate.setEnableTransactionSupport(true);

@Transactional(rollbackFor = Exception.class)
public String put() {
        int i = (int)(Math.random() * 100);
        template.opsForValue().set("key"+i, "value"+i, 300, TimeUnit.SECONDS);
        return "success "+"key"+i;
}
//方法中可以同时有Redis 和 SQL

我们可以先在 Spring 语境里配置一个 PlatformTransactionManager（例如 DataSourceTransactionManager），然后再用 @Transactional 注释来声明 Redis 事务的范围，让 Spring 自动关闭 Redis 连接。
另外，我们还发现了 Redis 事务和关系数据库事务（在本例中，即 JDBC）相结合的不利之处。混合型事务的表现和预想的不太一样。

方法2：无需@Transactional，手动执行事务（比较灵活），中间发生异常全部不执行（因为Redis事务是放入队列，Exec时一起执行）

redisTemplate.setEnableTransactionSupport(true);

无@Transactional ，手动配置执行事务，txResults 顺序的各行命令的执行结果

List<Object> txResults = redisTemplate.execute(new SessionCallback<List<Object>>() {
  public List<Object> execute(RedisOperations operations)  {
    operations.multi();
    operations.opsForSet().add("key", "value1");
    return operations.exec();
  }
});

事务建议

升级到springboot 2.0以上版本，如果因为项目原因无法升级看下面的建议
如果使用Redis事务的场景不多，完全可以自己管理(方法2)，不需要使用spring的注解式事务。

如果一定要使用spring提供的注解式事务，建议初始化两个RedisTemplate Bean，分别设置enableTransactionSupport属性为true和false。针对需要事务和不需要事务的操作使用不同的template。
从个人角度，我不建议使用redis事务，因为redis对于事务的支持并不是关系型数据库那样满足ACID。Redis事务只能保证ACID中的隔离性和一致性，无法保证原子性和持久性。而我们使用事务最重要的一个理由就是原子性，这一点无法保证，事务的意义就去掉一大半了。所以事务的场景可以尝试通过业务代码来实现。

pipeline管道化

使用Pipeline合并请求减少TCP链接次数。客户端允许将多个请求一次发给服务器，过程中而不需要等待请求的回复，在最后再一并读取结果即可。

pipeline机制可以优化吞吐量，但无法提供原子性/事务保障，而这个可以通过Redis-Multi等命令实现。
部分读写操作存在相关依赖，无法使用pipeline实现，可利用Script机制，但需要在可维护性方面做好取舍

https://www.cnblogs.com/littleatp/p/8419796.html

jedis操作
Pipeline pipeline = jedis.pipelined();
int j;
for (j = 0; j < batchSize; j++) {
     if (i + j < cmdCount) {
           pipeline.set(key(i + j), UUID.randomUUID().toString());
     } else {
           break;
     }
}
pipeline.sync(); //等待

执行LUA

https://blog.csdn.net/u014495560/article/details/82531046

串行化执行保证：原子性、有序性、可见性、减少通讯次数

Lua 脚本项目的 resources 目录下，起名 limit.lua 即可

注意：KEYS[1]，ARGV[]必须大写

local key = KEYS[1]; //必须大写
local dzyid=ARGV[1];

--转数字（对应key的值必须是redis数字才行，否则nil）
local current =tonumber( redis.call('get', key) ); 

--tonumber功能并不是将字符串转为数字，而是将redis数字转换LUA数字，为了在LUA中进行计算

if (current==1) then --数字比较
    return current+1;
end

//读取 lua 脚本
    @Bean
    public DefaultRedisScript<Number> redisluaScript() {
        DefaultRedisScript<Number> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("limit.lua")));
        redisScript.setResultType(Number.class); //返回类型
        return redisScript;
    }

    @Bean("simlockredisluaScriptLong")
    public DefaultRedisScript<Long> simlockredisluaScriptLong() {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lockdhhm.lua")));
        redisScript.setResultType(Long.class);//返回值类型 不要用Integer
        return redisScript;
    }

    @Bean
    public DefaultRedisScript<String> redisluaScriptStrig() {
        DefaultRedisScript<String> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("limit.lua")));
        redisScript.setResultType(String.class);//返回类型
        return redisScript;
    }

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
 
    @Autowired
    private DefaultRedisScript<Number> redisluaScript;

    List<String> keys= Arrays.asList("key1");
    //执行LUA  <T> T execute(RedisScript<T> var1, List<K> var2, Object... var3);
    Number number = redisTemplate.execute(redisluaScript,keys, arg1, arg2);

注意：序列化格式与存储格式、脚本返回类型匹配、Value类型与LUA数字计算、入参类型

一定要与redis中实际存储的数据类型匹配，否则会有各种异常

1 序列化：redisTemplate的序列化(String、Json、JDK序列化...)一定要与redis中Value保存的序列化格式一致，才能在get时正确的反序列化，否则会抛出异常。

2 返回值类型：T redisTemplate.execute返回值类型，取决于DefaultRedisScript设置的返回值类型。

3 lua的返回值类型： lua的返回值类型要与DefaultRedisScript设置的返回值类型完全匹配。比如value=1，返回值类型必须是Long,DefaultRedisScript设置为返回值String也会报错。所以可以设置返回值类型为Object，然后在通过instanceof判断类型

4 redis存储格式与LUA：注意设置数字时要这样执行redisTemplate.opsForValue().set("1",2) 。

不要这样执行redisTemplate.opsForValue().set("1","2")；

第二种设置的是字符串，LUA不支持对其tonumber()和运算，tonumber并不是将字符串转为数字

5 若LUA返回值为数字，要用Long/Number做JAVA的返回类型，不要用Integer；（会出现IllegalStateException）

6 参数类型一定和redis中匹配

常见异常报错

Caused by: io.lettuce.core.RedisCommandExecutionException: ERR Error running script (call to f_47e4c946b9a30705a03eded83a2e086b54791c44): @user_script:3: user_script:3: attempt to perform arithmetic on local 'current' (a nil value)

解释在lua脚本的第3行出错，local变量current是nil

SerializationException: Could not read JSON

序列化异常，要保证返回值格式符合redisTemplate解析格式。设置redisTemplate解析value为JSON，若LUA直接返回字符串需要加“”,否则无法解析JSON。

if (haskey == 0) then
    return '"失败"';
else
    return '"成功"';
end

java.lang.IllegalStateException

LUA返回值为整数时出现此异常，确定请在JAVA配置的返回值类型为Long而不是Integer。

LUA语法

注意：KEYS[1]，ARGV[]大写

1 数字相关的比较/计算必须要用tonumber()转化

if (tonumber(var)==tonumber(var1)) then

end

2 解析JSON（安装 lua-cjson 库） cjson.decode

local cjson = require "cjson"
local sampleJson = [[{"age":"23","testArray":{"array":[8,9,11,14,25]},"Himi":"himigame.com"}]];
--解析json字符串
local data = cjson.decode(sampleJson);
--打印json字符串中的age字段
print(data["age"]);
--打印数组中的第一个值(lua默认是从0开始计数)
print(data["testArray"]["array"][1]);

3 创建JSON （安装 lua-cjson 库） cjson.encode

local cjson = require "cjson"
local result ={}
result["@class"] = "org.meibaobao.ecos.basecomponent.common.Result"
result["success"] = true
result["data"] = current

cjson.encode(result);

4 字符串拼接..

'str1'..'str2'

5 判断redis.call('get',limitkey )返回null

redis没有get到key返回null，此时LUA中返回的结果不是 nil而是 userdata类型的 ngx.null。

当使用lua脚本执行逻辑时，如果要判断这个值，很容易让人迷惑以为它是nil，从而导致判断不成立，实际它是一个boolean的值

在lua中，除了nil和false，其他的值都为真，包括0，可以通过nil为false这一点来判断是否为空

local current = redis.call('get', key);
--在lua中，除了nil和false，其他的值都为真，包括0，可以通过nil为false这一点来判断是否为空
if current then
    return '"存在"';
else
    return '"不存在"';
end

一个简单LUA案例

模拟多人并发抢号的场景，抢到号码后锁定10分钟。10分钟后不操作释放号码。

redis中预热保存了50个号码测试信息，key为号码，value为号码信息。

--加载lua脚本    
@Bean("simlockredisluaScriptLong")
    public DefaultRedisScript<Long> simlockredisluaScriptLong() {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lockdhhm.lua")));
        redisScript.setResultType(Long.class);//返回值类型 不要用Integer
        return redisScript;
    }

@Autowired
@Qualifier("simlockredisluaScriptLong")
DefaultRedisScript<Long> simlockredisluaScriptLong;

int dhhm=111125730; //目标号码
String code="人员ID1";

--执行 注意参数的类型与redis匹配
Long result = redisTemplate.execute(simlockredisluaScriptLong, Arrays.asList(), dhhm, code);

LUA脚本代码 lockdhhm.lua

验证号码合法，验证个人锁定次数防刷，尝试加锁10分钟

--返回负值=失败
local key = 'sim:dhhm'; --号码列表
local dhhm = ARGV[1]; --目标号码
local dzyid = ARGV[2]; --员工ID
--local fdbs=ARGV[3]; --分店
local lockkey = 'sim:lock:' .. dhhm;

--检验号码已被锁（加速返回，不做后边的操作）
local belock = redis.call('get', lockkey);
if belock then
    --在lua中，除了nil和false，其他的值都为真，包括0，可以通过nil为false这一点来判断是否为空
    --已存在
    return -3;
end

--检验号码存在
if (redis.call('hexists', key, dhhm) == 0) then
    return -1;
end

--防刷 每个员工ID每个号码一小时只能锁3次
local limitkey = 'sim:limit:' .. dhhm .. ':' .. dzyid;

local limitnum = redis.call('get', limitkey);
if limitnum then
    if (tonumber(limitnum) > 2) then
        return -2;
    end
end

--尝试号码加锁 10分钟
local snkey = 'sim:sn:' .. dhhm;
if (redis.call('setnx', lockkey, dzyid) > 0) then
    redis.call('expire', lockkey, 600);

    --防刷 记录员工ID锁定此号次数
    if (redis.call('setnx', limitkey, 1) > 0) then
        --首次
        redis.call('expire', limitkey, 3600);
    else
        --累计
        redis.call('incr', limitkey);
    end

    --此序号用于携带给消费者，消费者核对当前消费是不是此号码最新加锁序号
    return redis.call('incr', snkey);
else
    return -3;
end

测试模拟多人并发抢同一号码

 --模拟多人并发抢同一号码

    @Autowired
    DefaultRedisScript<Long> simlockredisluaScriptLong;

    @Test
    public void simlock() throws InterruptedException {
        int num=100;
        int dhhm=111125730; //抢夺目标号码
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(50, 50,
                3, TimeUnit.MINUTES, new ArrayBlockingQueue<>(num));
        CountDownLatch begin = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(num);

        for (int i = 0; i < num; i++) {
            threadPoolExecutor.execute(() -> {
                try {
                    //测试 随机生成人员编号
                    String dzyid=String.valueOf(new Random().nextInt(100));
                    System.out.println(dzyid+"准备就绪");
                    begin.await();

                    Long result  = redisTemplate.execute(simlockredisluaScriptLong, Arrays.asList(), dhhm, dzyid);
                    if (result >=0){
                        System.out.println(dzyid+"成功锁定"+dhhm);
                    }
                    end.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        System.out.println("begin:"+System.currentTimeMillis());
        begin.countDown();
        end.await();
        System.out.println("end:"+System.currentTimeMillis());
    }

运行后显示 "84成功锁定111125730"，检查Redis数据库正确生成对应的锁