Redisson

背景

基于Redis(setnx)实现的分布式锁存在以下几个问题：

不可重入：同一个线程无法多次获取同一把锁
不可重试：获取锁只尝试一次就返回false，没有重试机制
超时释放：锁超时释放虽然可以避免死锁，但如果业务执行耗时较长，也会导致锁释放，存在一定安全隐患
主从一致性：如果Redis提供了主从集群，主从同步存在延迟，当主机宕机时，如果从机未同步主机中的锁数据，则会出现锁失效问题，存在一致性问题。

此时，如果自行编写代码去解决上述问题会显得业务逻辑十分复杂，所以，有没有别人已经封装好的工具，供我们直接使用呢？

此时，Redisson便呼之欲出了

简介

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

使用

既然知道Redisson可以解决一些分布式并发问题，那么该如何使用Redisson去解决生产开发中所遇到的问题呢？

这里提供两个链接进行学习：

https://github.com/redisson/redisson/wiki/目录(Redisson中文文档)
https://github.com/redisson/redisson(Redisson官方github)

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.20.0</version>
</dependency>

经历了9年多的版本迭代，最新的版本已经到3.20.0

配置类

在使用前需要为其创建一个配置类(RedisConfig)

@Configuration
public class RedisConfig {
	@Bean
	public RedissonClient redissonClient(){
		// 配置类
		Config config = new Config();
		// 添加redis地址，这里添加了单点的地址，也可以使用config.useClusterServers()添加集群地址
		config.useSingleServer().setAddress("redis://127.0.0.1:6379");
		// 添加客户端
		return Redisson.create(config);
	}
}

将其通过bean的方式注入进spring，后续使用时直接利用@Autowried或@Resource完成自动装配进行使用。

源代码

通过跟踪查看源代码，我们可以了解到Redisson底层是如何分别解决上面的几个问题的

解决不可重入问题：利用Hash结构记录线程id和重入次数

添加锁源码

释放锁源码

从上面的源码中可以得知，Redisson是通过一个Lua脚本来解决Redis中不可重入问题的，具体代码逻辑如下图

解决不可重试问题：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制

@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return true;
    }

    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    current = System.currentTimeMillis();
    RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
        if (!subscribeFuture.cancel(false)) {
            subscribeFuture.onComplete((res, e) -> {
                if (e == null) {
                    unsubscribe(subscribeFuture, threadId);
                }
            });
        }
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    try {
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            // waiting for message
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(subscribeFuture, threadId);
    }
    //        return get(tryLockAsync(waitTime, leaseTime, unit));
}

通过源码跟踪，Redisson在获取锁的时候利用了消息订阅和信号量的机制(传递了等待重试时间)，并不是无休止的盲等机制。

解决超时释放问题：利用watchDog(看门狗)，每隔一段时间(releaseTime / 3),重置超时时间

源码

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                                         commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                         TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }

        // lock acquired
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}