Wie Springboot die auf Redisson basierende verteilte reentrant Lock-Quellcodeanalyse von Redis implementiert

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>
<!--redis分布式锁-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.12.0</version>
</dependency>

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author wangzhenjun
 * @date 2022/2/9 9:57
 */
@Configuration
public class MyRedissonConfig {

    /**
     * 所有对redisson的使用都是通过RedissonClient来操作的
     * @return
     */
    @Bean(destroyMethod="shutdown")
    public RedissonClient redisson(){
        // 1. 创建配置
        Config config = new Config();
        // 一定要加redis://
        config.useSingleServer().setAddress("redis://192.168.17.130:6379");
        // 2. 根据config创建出redissonClient实例
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

@ResponseBody
@GetMapping("/hello")
public String hello(){
    // 1.获取一把锁，只要锁名字一样，就是同一把锁
    RLock lock = redisson.getLock("my-lock");
    // 2. 加锁
    lock.lock();// 阻塞试等待  默认加的都是30s
    // 带参数情况
    // lock.lock(10, TimeUnit.SECONDS);// 10s自动解锁，自动解锁时间一定要大于业务的执行时间。
    try {
        System.out.println("加锁成功" + Thread.currentThread().getId());
        Thread.sleep(30000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        // 3. 解锁
        System.out.println("解锁成功：" + Thread.currentThread().getId());
        lock.unlock();
    }
    return "hello";
}

@Override
public void lock() {
    try {
    	// 我们发现不穿过期时间源码默认过期时间为-1
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
	// 获取线程的id，占有锁的时候field的值为UUID:线程号id
    long threadId = Thread.currentThread().getId();
    // 尝试获得锁
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired 获得锁，返回
    if (ttl == null) {
        return;
    }
	// 这里说明获取锁失败，就通过线程id订阅这个锁
    RFuture<RedissonLockEntry> future = subscribe(threadId);
    if (interruptibly) {
        commandExecutor.syncSubscriptionInterrupted(future);
    } else {
        commandExecutor.syncSubscription(future);
    }

    try {
    	// 这里进行自旋，不断尝试获取锁
        while (true) {
        	// 继续尝试获取锁
            ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired 获取成功
            if (ttl == null) {
            	// 直接返回，挑出自旋
                break;
            }

            // waiting for message 继续等待获得锁
            if (ttl >= 0) {
                try {
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
            } else {
                if (interruptibly) {
                    future.getNow().getLatch().acquire();
                } else {
                    future.getNow().getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
     	// 取消订阅
        unsubscribe(future, threadId);
    }
//        get(lockAsync(leaseTime, unit));
}

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
	// 直接进入异步方法
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    // 这里进行判断如果没有设置参数leaseTime = -1
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 此方法进行获得锁，过期时间为看门狗的默认时间
    // private long lockWatchdogTimeout = 30 * 1000;看门狗默认过期时间为30s
    // 加锁和过期时间要保证原子性，这个方法后面肯定调用执行了Lua脚本，我们下面在看
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(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;

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }
    
    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {
            ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
            if (ent == null) {
                return;
            }
            Long threadId = ent.getFirstThreadId();
            if (threadId == null) {
                return;
            }
            
            RFuture<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> {
                if (e != null) {
                    log.error("Can&#39;t update lock " + getName() + " expiration", e);
                    return;
                }
                
                if (res) {
                    // reschedule itself
                    renewExpiration();
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
    ee.setTimeout(task);
}

@Override
public void lock(long leaseTime, TimeUnit unit) {
    try {
    	// 这里的过期时间为我们输入的10
        lock(leaseTime, unit, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    // 这里进行判断如果没有设置参数leaseTime = -1,此时我们为10
    if (leaseTime != -1) {
    	// 来到此方法
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 此处省略后面内容，前面以详细说明。。。。
}

@Override
public void unlock() {
    try {
    	// 点击进入释放锁方法
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        if (e.getCause() instanceof IllegalMonitorStateException) {
            throw (IllegalMonitorStateException) e.getCause();
        } else {
            throw e;
        }
    }
    
//        Future<Void> future = unlockAsync();
//        future.awaitUninterruptibly();
//        if (future.isSuccess()) {
//            return;
//        }
//        if (future.cause() instanceof IllegalMonitorStateException) {
//            throw (IllegalMonitorStateException)future.cause();
//        }
//        throw commandExecutor.convertException(future);
}

@Override
public RFuture<Void> unlockAsync(long threadId) {
    RPromise<Void> result = new RedissonPromise<Void>();
    // 解锁方法，后面展开说
    RFuture<Boolean> future = unlockInnerAsync(threadId);
	// 完成
    future.onComplete((opStatus, e) -> {
        if (e != null) {
        	// 取消到期续订
            cancelExpirationRenewal(threadId);
            // 将这个未来标记为失败并通知所有人
            result.tryFailure(e);
            return;
        }
		// 状态为空，说明解锁的线程和当前锁不是同一个线程
        if (opStatus == null) {
            IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + threadId);
            result.tryFailure(cause);
            return;
        }
        
        cancelExpirationRenewal(threadId);
        result.trySuccess(null);
    });

    return result;
}