Redisson相关使用&&分布式锁浅析

2023-05-16

注：本篇的redisson版本基于3.13.3；本篇的demo将我写的源代码贴了出来，每个方法都有清晰的注释，分布式锁相关的代码以及验证是我手动验证Redis中key状态来判断的。

文章目录

- 简介
- Redisson配置
- Redisson的对象相关操作
- Redisson集合操作
- 分布式锁相关
- - Redisson核心lua操作代码及步骤
- 其他
- 参考资料
- 本篇源代码

简介

Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

简单来说好处就是，可以在写Java代码时，通过Java的类来操作存储。比如hash可以直接转成Map。redission也是为分布式环境提供帮助的Redis框架。提供了分布式锁的实现。简单容易。redisson支持分片集群存储。

Redisson的操作是基于Redis的hash数据结构，创建和删除key作为加锁和解锁的操作，hash的value中的数字次数作为重入的次数。

Redisson配置

本篇是基于springboot的配置

首先是pom.xml的jar包引入，下面的jar是本次需要加的

<dependency>
           <groupId>org.redisson</groupId>
           <artifactId>redisson</artifactId>
           <version>${redisson.version}</version>
</dependency>
<dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-configuration-processor</artifactId>
           <optional>true</optional>
</dependency>

然后是application.yml

redisson:
  enable: true
  database: 1
  password: 12345678
  address: redis://127.0.0.1:6379
  connectTimeout: 5000
  pingConnectionInterval: 5000
  timeout: 5000

然后使用@ConfigurationProperties

package com.zy.integrate.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * @author zhangyong05
 * Created on 2021-02-24
 */
@Component
@ConfigurationProperties(prefix="redisson",ignoreInvalidFields = false)
public class RedissonProperties {
    // 加setter&&getter
    private Integer database;
    private String password;
    private String address;
    private Integer connectTimeout;
    private Integer pingConnectionInterval;
    private Integer timeout;
}

----------------------------------------------
package com.zy.integrate.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * @author zhangyong05
 * Created on 2021-02-24
 */
@Component
@ConditionalOnProperty(prefix = "redisson",value = "enable", havingValue = "true")
@EnableConfigurationProperties(RedissonProperties.class)
public class RedissonConfig {

    /**
     * 有 ConfigurationProperties 和 Configuration && Value两种注入配置文件方式
     * ConfigurationProperties 更加方便
     */
    @Autowired
    private RedissonProperties redissonProperties;

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress(redissonProperties.getAddress())
                .setDatabase(redissonProperties.getDatabase())
                .setPassword(redissonProperties.getPassword())
                .setConnectTimeout(redissonProperties.getConnectTimeout())
                .setPingConnectionInterval(redissonProperties.getPingConnectionInterval())
                .setTimeout(redissonProperties.getTimeout());
        return Redisson.create(config);
    }
}

以上redisson就配置好了

Redisson的对象相关操作

package com.zy.integrate.controller;

import com.zy.integrate.domain.Persion;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author zhangyong05
 * Created on 2021-02-24 
 * https://www.javadoc.io/doc/org.redisson/redisson/3.10.3/org/redisson/api/RTopic.html
 * 可以参考： https://github.com/redisson/redisson-examples
 */
@RestController
public class RedissonObjectController {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 获取keys && 根据pattern 获取keys
     * @param 
     * @return  
     * @author zhangyong05 
     * 2021/2/26 
     */
    @GetMapping("/redisson-key")
    public void keyTest(){
        RKeys keys = redissonClient.getKeys();
        Iterable<String> keysByPattern = keys.getKeysByPattern("s*");
        for (String s : keysByPattern) {
            System.out.println(s);
        }
    }

    /**
     * 验证redis 的string类型操作
     * @author zhangyong05 
     * 2021/2/26 
     */
    @GetMapping("/redisson-string")
    public String stringTest(){
        // https://stackoverflow.com/questions/51276593/whats-the-usage-for-tryset-method-in-redisson-rbucket
        RBucket<String> bucket = redissonClient.getBucket("string-test");
        // trySet当value为空时会设置成功
        boolean trySetValue = bucket.trySet("trySetValue");
        System.out.println(trySetValue);
        String res = bucket.get();
        bucket.compareAndSet("except","update");
        String before = bucket.getAndSet("after");
        // size返回的是对象的大小所占字节，并非是长度。
        System.out.println(bucket.size());
        System.out.println(before);
        bucket.set(System.currentTimeMillis() + "asd");
        return res;
    }

    /**
     * 验证 Redis的 string存储自定义对象的操作,需要注意的是redisson的codec间接决定了能否存储对象，以及编解码方式
     * codec编解码配置需要一致才能正常序列化和反序列化
     * @author zhangyong05 
     * 2021/2/26 
     */
    @GetMapping("/redisson-object")
    public Persion objectTest(){
        // redisson的默认编码codec支持对象存Redis的string类型里面
        Persion persion = new Persion();
        persion.setName("张三");
        persion.setTime(System.currentTimeMillis());
        persion.setAge(18);
        RBucket<Persion> bucket = redissonClient.getBucket("object-test");
        Persion res = bucket.get();
        bucket.set(persion);
        return res;
    }

    /**
     * 验证原子类，也是Redis中的string类型存储
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-atomic-long")
    public Long atomicLongTest(){
        RAtomicLong atomicLong = redissonClient.getAtomicLong("atomic-long-test");
        atomicLong.addAndGet(1);
        atomicLong.addAndGet(1);
        return atomicLong.get();
    }

    /**
     * 验证发布订阅，调用这个接口就能触发，因为写了一个TopicService类在程序启动时去运行监听topic订阅方
     * 这个接口是用来publish消息的
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-topic")
    public void topicTest() {
        RTopic topic = redissonClient.getTopic("test-topic");
        // 程序启动时有ApplicationRunner实现类注册监听器：topicListener
        topic.publish("msg");
        // 也可以用topic模式，同一模式下都会收到消息
    }

}

Redisson集合操作

package com.zy.integrate.controller;

import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * @author zhangyong05
 * Created on 2021-02-25
 */
@RestController
public class RedissonCollectionController {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 对应Redis中的hash 还有cache以及local以及multi多种操作方式。但是我觉得使用场景并不多就没写demo
     * cache是给value加了 ttl
     * local是加了本地缓存
     * multi是提供多操作，Java对象允许Map中的一个字段值包含多个元素。
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-map")
    public void mapTest(){
        // 对应 HSET field value
        RMap<String, String> map = redissonClient.getMap("map-test");
        map.putIfAbsent("field","value1");
        map.put("123","123qwe");
        map.fastPut("123","456rty");

        // mapCache 可以设置value的消失时间ttl 以及 最长闲置时间 挺鸡肋的,感觉作用不大，性能下滑。
        RMapCache<String, String> mapCache = redissonClient.getMapCache("cache-map-test");
        mapCache.put("sad","eqw",1, TimeUnit.HOURS,1,TimeUnit.MINUTES);
    }

    /**
     * 对应Redis中的set，可以排序
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-set")
    public void setTest(){
        RSet<String> set = redissonClient.getSet("set-test");
        set.add("asd");
        set.add("123");
        set.add("456");
        set.add("789");
        set.remove("456");

        // set也可以用cache 还有批量操作，但是感觉这种功能使用场景比较少。
        // 排序的sort 可以指定comparator
        RSortedSet<Integer> sortedSet = redissonClient.getSortedSet("sort-set-test");
        sortedSet.add(6);
        sortedSet.add(3);
        sortedSet.add(1);
        sortedSet.add(7);
    }

    /**
     * 带有score的zset操作
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-zset")
    public void setScoreTest(){
        RScoredSortedSet<String> scoredSortedSet = redissonClient.getScoredSortedSet("score-set-test");
        scoredSortedSet.add(90.22,"数学");
        scoredSortedSet.add(98.22,"语文");
        scoredSortedSet.add(92.22,"英语");
        // 相同的覆盖，最后英语为93.22
        scoredSortedSet.add(93.22,"英语");
        Double score = scoredSortedSet.getScore("数学");
        System.out.println(score);
        // rank 从0起始
        Integer rank = scoredSortedSet.rank("数学");
        System.out.println(rank);
    }

    /**
     * 就是一个放到Redis中的队列  list
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-queue")
    public void queueTest(){
        // 无界队列
        RQueue<String> queue = redissonClient.getQueue("queue-test");
        // queue 中使用offer和poll 做入队和出队的操作。poll会删除掉队首元素
        queue.offer("sad");
        queue.offer("wqe");
        queue.offer("123");
        queue.offer("456");
        queue.poll();
    }

    /**
     * 有界队列 list 这个可以作为过滤进几次筛选结果的需求等等
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-bound-queue")
    public void boundBlockQueue(){
        // 使用场景过滤近几次筛选过的结果
        RBoundedBlockingQueue<Long> boundedBlockingQueue = redissonClient.getBoundedBlockingQueue("bound-queue-test");
        // 设置有界队列的长度为2
        int bound = 2;
        boundedBlockingQueue.trySetCapacity(bound);
        // offer操作，当队列满时就不加了；判断队列满的话先出队再入队；
        if (boundedBlockingQueue.size() == bound){
            boundedBlockingQueue.poll();
        }
        // 可以验证Redis的值
        boundedBlockingQueue.offer(System.currentTimeMillis());
    }

    /**
     * 优先队列，list 这个也可以作为不断取最高(低)的 这种需求
     * @param
     * @return
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-priority-queue")
    public void priorityQueueTest(){
        // 优先队列，可以设置comparator
        RPriorityQueue<Integer> priorityQueue = redissonClient.getPriorityQueue("priority-queue-test");
        priorityQueue.offer(3);
        priorityQueue.offer(1);
        priorityQueue.offer(4);
        System.out.println(priorityQueue.poll());
    }

    /**
     * 双端优先队列
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-priority-deque")
    public void priorityDequeTest(){
        // LIST
        RPriorityDeque<Integer> priorityDeque = redissonClient.getPriorityDeque("priority-deque-test");
        priorityDeque.add(3);
        priorityDeque.add(1);
        priorityDeque.add(4);
        // 当队列没有数据时，poll操作不会报错 如果触发进俩出去俩，队列为空，Redis不展示这个key
        priorityDeque.pollLast();
        priorityDeque.pollFirst();
    }

}

分布式锁相关

package com.zy.integrate.controller;

import org.redisson.api.*;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangyong05
 * Created on 2021-02-25
 */
@RestController
public class RedissonLockController {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 分布式锁，这个接口验证的是可重入锁，跟ReentrantLock类似，Redis会记录重入次数value
     * 当释放的时候如果重入先 value-- 否则删除key
     * demo设置了10s等待锁时间，60s加锁释放时间，即使重入也60s删除
     * (问题是超过60s就释放锁不管是否业务执行完毕，所以需要控制回滚事务---后面有watchDog方式)
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-lock")
    public String lockTest(){
        System.out.println("请求触发时间: "+ LocalDateTime.now());
        RLock lock = redissonClient.getLock("test-lock-key");
        try {
            // 10是waitTime等待时间  60是 leaseTime 加锁时间，如果unlock就提前释放锁
            // 如果宕机的话，先是finally会释放锁(finally一般情况下是没问题的,可能个别极端情况有问题，我还没遇到过可以验证)，如果没释放成功的话，就leaseTime后自动释放锁。
            boolean tryLock = lock.tryLock(10, 60, TimeUnit.SECONDS);
            // tryLock执行完就加完锁了 如果返回false就加锁失败，跟Lock一样
            if (!tryLock){
                return "请稍后再试";
            }
            boolean reentrantLock = lock.tryLock(10, 60, TimeUnit.SECONDS);
            if (!reentrantLock){
                return "可重入锁失败！";
            }
            // 验证可重入锁释放在业务执行完成之前，之后再unlock就抛异常了
            Thread.sleep(75000);
            // todo something 一些有竞争的业务代码
            // 释放重入锁，value--;
            lock.unlock();
            System.out.println("释放重入锁时间: "+ LocalDateTime.now());
            // 验证锁删除在业务执行完成之前
            Thread.sleep(75000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            // 释放锁如果当前只有一个锁（非重入状态），会把这个hash key删掉test-lock-key
            System.out.println("释放lock时间: "+ LocalDateTime.now());
            lock.unlock();
        }
        return "success";
    }

    /**
     * 分布式锁，watchDog方式，watchDog在没指定 leaseTime的时候会触发
     * LockWatchdogTimeout控制锁的存在时间，LockWatchdogTimeout/3是检测锁续期的周期。
     * 开启watchDog会保证业务逻辑执行完成之后才释放锁。不断的检测续期。
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-watch-dog")
    public String watchDogTest(){
        // watchDog的延时时间 可以由 lockWatchdogTimeout指定默认延时时间，默认30s
        // watchDog 只有在未显示指定加锁时间时(leaseTime)才会生效；watchDog的效果是不释放锁，延长锁的时间防止并发问题.
        // watchDog 在当前线程没有执行结束的情况下，会每lockWatchdogTimeout/3时间，去检测延时，会重新设置timeout时间为30s（即刷新时间）；
        System.out.println("请求触发时间: "+ LocalDateTime.now());
        Config config = redissonClient.getConfig();
        // LockWatchdogTimeout: 30000
        // 可以验证，当调用这个接口时，把这个程序关停掉，发现Redis key的删除时间和触发时间正好LockWatchdogTimeout时长相等
        // (如果程序是触发接口10s钟后关掉的，可能触发了延期，那么就是从最近延期那个时间为起点的30s会删除key)
        // 例如 11:30调用接口，11:45关闭程序，那么key在12:15删除
        System.out.println("LockWatchdogTimeout: "+config.getLockWatchdogTimeout());
        RLock lock = redissonClient.getLock("watch-dog-test");
        try {
            // 这个时间跟reentrantLock的一致，是等待锁的时间，并非加锁时间leaseTime
            // watchDog开启时会消耗性能，可以设置leaseTime给业务执行时间，意外超时就事务回滚
            boolean tryLock = lock.tryLock(10, TimeUnit.SECONDS);
            if (!tryLock){
                return "请稍后再试";
            }
            // 模拟业务耗时,当前没有显示指定时间，默认时间是30s释放锁,通过查看Redis的key可以看到key是在73s的时候删除的(unlock)
            // 因此可以验证watchDog进行了续期（debug不可以验证，请用耗时操作例如：Thread.sleep）
            Thread.sleep(73000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("释放最后锁的时间"+LocalDateTime.now());
            lock.unlock();
        }
        return "success";
    }

    /**
     * 这个接口验证watchDog在重入时的生效效果；
     * 可以看到当重入锁释放时value进行了减1的操作，key删除是在finally执行完删除的.
     * 效果是只有业务代码没执行完就不会删除掉锁
     * 请求触发时间: 2021-02-26T20:50:54.180
     * 释放重入锁的时间2021-02-26T20:52:07.201
     * 释放lock的时间2021-02-26T20:53:20.210
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-watch-dog-2")
    public String watchDogReentrantTest(){
        System.out.println("请求触发时间: "+ LocalDateTime.now());
        RLock lock = redissonClient.getLock("watch-dog-test2");
        try {
            boolean tryLock = lock.tryLock(10, TimeUnit.SECONDS);
            if (!tryLock){
                return "请稍后再试";
            }
            boolean reentrantLock = lock.tryLock(10, TimeUnit.SECONDS);
            if (!reentrantLock){
                return "请稍后再试,重入锁！";
            }
            Thread.sleep(73000);
            System.out.println("释放重入锁的时间"+LocalDateTime.now());
            lock.unlock();
            Thread.sleep(73000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            System.out.println("释放lock的时间"+LocalDateTime.now());
            lock.unlock();
        }
        return "success";
    }

    /**
     * 事务处理 事务底层是通过lua实现的
     * 如下代码中 int a= 1/0 这行注释解开就能验证回滚
     * @author zhangyong05
     * 2021/2/26
     */
    @GetMapping("/redisson-transaction")
    public void transactionTest(){
        RTransaction transaction = redissonClient.createTransaction(TransactionOptions.defaults());
        try {
            RMap<String, String> map = transaction.getMap("transaction-test");
            map.put("123","4");
            map.put("456","5");
            // 开启下面注释就抛异常就会回滚事务
//            int a = 1/0;
            map.put("567","6");
            transaction.commit();
        }catch (Exception e){
            transaction.rollback();
        }
    }
}

Redisson核心lua操作代码及步骤

Redisson加锁的lua脚本

"if (redis.call('exists', KEYS[1]) == 0) 
then 
redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); 
return nil; 
end; 
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); 
return nil; 
end; 
return redis.call('pttl', KEYS[1]);"

KEYS[1] 代表加锁的key

ARGV[1] 代表的是锁key 的默认生存时间，leaseTime

ARGV[2] 代表的是加锁的客户端 UUID+Thread_ID

加锁流程：

判断key是否存在，如果不存在就给这个hash类型的key设置值；key field value 其中key是传入的key，field是客户端UUID+Thread_ID， Value 是 1；并设置锁的消失时间为leaseTime
第二个if判断，如果当前key下的客户端存在，那么就value+1 重入；
最后一行是代表其他线程进入，那么就会返回锁的剩余时间，不进行别的操作；

Ression解锁的lua脚本

"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) 
then 
return nil;
end; 
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); 
if (counter > 0) then 
  redis.call('pexpire', KEYS[1], ARGV[2]);
  return 0; 
else 
  redis.call('del', KEYS[1]); 
  redis.call('publish', KEYS[2], ARGV[1]); 
  return 1; 
end; 
return nil;"

KEYS[1] 是key

KEYS[2] 是getChannelName()，即KEYS[2] = redisson_lock__channel:{xxx}

ARGV[3] 是field

ARGV[2] 是leaseTime

ARGV[1] 是LockPubSub.unlockMessage，即ARGV[1] =0

解锁流程:

如果key filed不存在就返回nil，说明锁可用；
否则执行判断且value减1，计算结果counter
如果counter > 0 证明是可重入锁，重新设置过期时间
如果counter <= 0 则可以释放了，删除这个key 并发布 topic同步给阻塞的线程去获取锁。

其他

Redisson作为分布式锁在分布式开发的时候很常见，一般我们都以业务的某个唯一key为鉴别条件。例如：购买商品，给这个商品ID为Redis的key上分布式锁，然后所有用户对这个商品库存的操作就会按顺序执行而非竞争。我们常规使用的synchronized和ReentrantLock是本地的锁，因为分布式都是多实例，这种无法达到原子一致效果。因此有了分布式锁的出现。Redisson的分布式锁是以hash 数据结构为准的，并且支持重入。

hash   key  field  value
其中field是UUID+Thread-ID 
value是数字 1...    控制重入次数

参考资料

方法API手册
GitHub例子Demo
Redisson文档中文翻译
https://zhuanlan.zhihu.com/p/135864820

本篇源代码

https://github.com/StrandingHeart/JavaIntegration/tree/feature/redisson

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)