分布式锁—3.Redisson的公平锁

大纲
1.Redisson公平锁RedissonFairLock概述
2.公平锁源码之加锁和排队
3.公平锁源码之可重入加锁
4.公平锁源码之新旧版本对比
5.公平锁源码之队列重排
6.公平锁源码之释放锁
7.公平锁源码之按顺序依次加锁
 
1.Redisson公平锁RedissonFairLock概述
(1)非公平和公平的可重入锁
(2)Redisson公平锁的简单使用
(3)Redisson公平锁的初始化
 
(1)非公平和公平的可重入锁
一.非公平可重入锁
锁被释放后，排队获取锁的线程会重新无序获取锁，没有任何顺序性可言。
 
二.公平可重入锁
锁被释放后，排队获取锁的线程会按照请求获取锁时候的顺序去获取锁。公平锁可以保证线程获取锁的顺序，与其请求获取锁的顺序是一样的。也就是谁先申请获取到这把锁，谁就可以先获取到这把锁。公平可重入锁会把各个线程的加锁请求进行排队处理，保证先申请获取锁的线程，可以优先获取锁，从而实现所谓的公平性。
 
三.可重入的非公平锁和公平锁不同点
可重入的非公平锁和公平锁，在整体的技术实现框架上都是一样的。唯一的不同点就是加锁和解锁的逻辑不一样。非公平锁的加锁逻辑，比较简单。公平锁的加锁逻辑，要加入排队机制，保证各个线程排队能按顺序获取锁。
 
(2)Redisson公平锁的简单使用
Redisson的可重入锁RedissonLock指的是非公平可重入锁，Redisson的公平锁RedissonFairLock指的是公平可重入锁。
 
Redisson的公平可重入锁实现了java.util.concurrent.locks.Lock接口，保证了当多个线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒之后才会继续分配下一个线程。
 
RedissonFairLock是RedissonLock的子类。RedissonFairLock的锁实现框架，和RedissonLock基本一样。而在获取锁和释放锁的lua脚本中，RedissonFairLock的逻辑才有所区别。

1. //1.最常见的使用方法
2. RedissonClient redisson = Redisson.create(config);
3. RLock fairLock = redisson.getFairLock("myLock");
4. fairLock.lock();
6. //2.10秒钟以后自动解锁，无需调用unlock方法手动解锁
7. fairLock.lock(10, TimeUnit.SECONDS);
9. //3.尝试加锁，最多等待100秒，上锁以后10秒自动解锁
10. boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
11. fairLock.unlock();
13. //4.Redisson为公平的可重入锁提供了异步执行的相关方法
14. RLock fairLock = redisson.getFairLock("myLock");
15. fairLock.lockAsync();
16. fairLock.lockAsync(10, TimeUnit.SECONDS);
17. Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);

复制代码

(3)Redisson公平锁的初始化

1. public class RedissonDemo {
2.     public static void main(String[] args) throws Exception {
3.         ...
4.         //创建RedissonClient实例
5.         RedissonClient redisson = Redisson.create(config);
6.         
7.         //获取公平的可重入锁
8.         RLock fairLock = redisson.getFairLock("myLock");
9.         fairLock.lock();//加锁
10.         fairLock.unlock();//释放锁
11.     }
12. }
14. public class Redisson implements RedissonClient {
15.     //Redis的连接管理器，封装了一个Config实例
16.     protected final ConnectionManager connectionManager;
17.     //Redis的命令执行器，封装了一个ConnectionManager实例
18.     protected final CommandAsyncExecutor commandExecutor;
19.     ...
20.     protected Redisson(Config config) {
21.         this.config = config;
22.         Config configCopy = new Config(config);
23.         //初始化Redis的连接管理器
24.         connectionManager = ConfigSupport.createConnectionManager(configCopy);
25.         ...  
26.         //初始化Redis的命令执行器
27.         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
28.         ...
29.     }
30.    
31.     public RLock getFairLock(String name) {
32.         return new RedissonFairLock(commandExecutor, name);
33.     }
34.     ...
35. }
37. public class RedissonFairLock extends RedissonLock implements RLock {
38.     private final long threadWaitTime;
39.     private final CommandAsyncExecutor commandExecutor;
40.     ...
41.     public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name) {
42.         this(commandExecutor, name, 60000*5);
43.     }
44.    
45.     public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name, long threadWaitTime) {
46.         super(commandExecutor, name);
47.         this.commandExecutor = commandExecutor;
48.         this.threadWaitTime = threadWaitTime;
49.          ...
50.     }
51.     ...
52. }
54. public class RedissonLock extends RedissonBaseLock {
55.     protected long internalLockLeaseTime;
56.     final CommandAsyncExecutor commandExecutor;
57.     ...
58.     public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
59.         super(commandExecutor, name);
60.         this.commandExecutor = commandExecutor;
61.         //与WatchDog有关的internalLockLeaseTime
62.         //通过命令执行器CommandExecutor可以获取连接管理器ConnectionManager
63.         //通过连接管理器ConnectionManager可以获取Redis的配置信息类Config
64.         //通过Redis的配置信息类Config可以获取lockWatchdogTimeout超时时间
65.         this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
66.         ...
67.     }
68.     ...
69. }
71. public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {
72.     ...
73.     protected long internalLockLeaseTime;
74.     final String id;
75.     final String entryName;
76.     final CommandAsyncExecutor commandExecutor;
77.    
78.     public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
79.         super(commandExecutor, name);
80.         this.commandExecutor = commandExecutor;
81.         this.id = commandExecutor.getConnectionManager().getId();//获取UUID
82.         this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
83.         this.entryName = id + ":" + name;
84.     }
85.     ...
86. }
88. abstract class RedissonExpirable extends RedissonObject implements RExpirable {
89.     RedissonExpirable(CommandAsyncExecutor connectionManager, String name) {
90.         super(connectionManager, name);
91.     }
92.     ...
93. }
95. public abstract class RedissonObject implements RObject {
96.     protected final CommandAsyncExecutor commandExecutor;
97.     protected String name;
98.     protected final Codec codec;
99.    
100.     public RedissonObject(CommandAsyncExecutor commandExecutor, String name) {
101.         this(commandExecutor.getConnectionManager().getCodec(), commandExecutor, name);
102.     }
103.    
104.     public RedissonObject(Codec codec, CommandAsyncExecutor commandExecutor, String name) {
105.         this.codec = codec;
106.         this.commandExecutor = commandExecutor;
107.         if (name == null) {
108.             throw new NullPointerException("name can't be null");
109.         }
110.         setName(name);
111.     }
112.     ...
113. }
115. public class ConfigSupport {
116.     ...
117.     //创建Redis的连接管理器
118.     public static ConnectionManager createConnectionManager(Config configCopy) {
119.         //生成UUID
120.         UUID id = UUID.randomUUID();
121.         ...
122.         if (configCopy.getClusterServersConfig() != null) {
123.             validate(configCopy.getClusterServersConfig());
124.             //返回ClusterConnectionManager实例
125.             return new ClusterConnectionManager(configCopy.getClusterServersConfig(), configCopy, id);
126.         }
127.         ...
128.     }
129.     ...
130. }
132. public class ClusterConnectionManager extends MasterSlaveConnectionManager {
133.     public ClusterConnectionManager(ClusterServersConfig cfg, Config config, UUID id) {
134.         super(config, id);
135.         ...
136.     }
137.     ...
138. }
140. public class MasterSlaveConnectionManager implements ConnectionManager {
141.     protected final String id;//初始化时为UUID
142.     private final Config cfg;
143.     protected Codec codec;
144.     ...
145.     protected MasterSlaveConnectionManager(Config cfg, UUID id) {
146.         this.id = id.toString();//传入的是UUID
147.         ...
148.         this.cfg = cfg;
149.         this.codec = cfg.getCodec();
150.         ...
151.     }
152.    
153.     public String getId() {
154.         return id;
155.     }
156.    
157.     public Codec getCodec() {
158.         return codec;
159.     }
160.     ...
161. }

复制代码

 
2.公平锁源码之加锁和排队
(1)加锁时的执行流程
(2)获取公平锁的lua脚本相关参数说明
(3)lua脚本步骤一：进入while循环移除队列和有序集合中等待超时的线程
(4)lua脚本步骤二：判断当前线程能否获取锁
(5)lua脚本步骤三：执行获取锁的操作
(6)lua脚本步骤四：判断锁是否已经被当前线程持有(可重入锁)
(7)lua脚本步骤五：判断当前获取锁失败的线程是否已经在队列中排队
(8)lua脚本步骤六：对获取锁失败的线程进行排队
(9)获取锁失败的第一个线程执行lua脚本的流程
(10)获取锁失败的第二个线程执行lua脚本的流程
 
(1)加锁时的执行流程
使用Redisson的公平锁RedissonFairLock进行加锁时：首先调用的是RedissonLock的lock()方法，然后会调用RedissonLock的tryAcquire()方法，接着会调用RedissonLock的tryAcquireAsync()方法。
 
在RedissonLock的tryAcquireAsync()方法中，会调用一个可以被RedissonLock子类重载的tryLockInnerAsync()方法。对于非公平锁，执行到这会调用RedissonLock的tryLockInnerAsync()方法。对于公平锁，执行到这会调用RedissonFairLock的tryLockInnerAsync()方法。
 
在RedissonFairLock的tryLockInnerAsync()方法中，便执行具体的lua脚本。
[code]public class RedissonDemo {    public static void main(String[] args) throws Exception {        ...        //创建RedissonClient实例        RedissonClient redisson = Redisson.create(config);                //获取公平的可重入锁        RLock fairLock = redisson.getFairLock("myLock");        fairLock.lock();//加锁        fairLock.unlock();//释放锁    }}public class RedissonLock extends RedissonBaseLock {    ...    //不带参数的加锁    public void lock() {        try {            lock(-1, null, false);        } catch (InterruptedException e) {            throw new IllegalStateException();        }    }        //带参数的加锁    public void lock(long leaseTime, TimeUnit unit) {        try {            lock(leaseTime, unit, false);        } catch (InterruptedException e) {            throw new IllegalStateException();        }    }        private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {        long threadId = Thread.currentThread().getId();        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);        //加锁成功        if (ttl == null) {            return;        }        //加锁失败        ...    }        private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {        return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));    }        private  RFuture tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {        RFuture ttlRemainingFuture;        if (leaseTime != -1) {            ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);        } else {            //非公平锁，接下来调用的是RedissonLock.tryLockInnerAsync()方法            //公平锁，接下来调用的是RedissonFairLock.tryLockInnerAsync()方法            ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);        }        //对RFuture类型的ttlRemainingFuture添加回调监听        CompletionStage f = ttlRemainingFuture.thenApply(ttlRemaining -> {            //tryLockInnerAsync()里的加锁lua脚本异步执行完毕，会回调如下方法逻辑：            //加锁成功            if (ttlRemaining == null) {                if (leaseTime != -1) {                    //如果传入的leaseTime不是-1，也就是指定锁的过期时间，那么就不创建定时调度任务                    internalLockLeaseTime = unit.toMillis(leaseTime);                } else {                    //创建定时调度任务                    scheduleExpirationRenewal(threadId);                }            }            return ttlRemaining;        });        return new CompletableFutureWrapper(f);    }    ...}public class RedissonFairLock extends RedissonLock implements RLock {    private final long threadWaitTime;//线程可以等待锁的时间    private final CommandAsyncExecutor commandExecutor;    private final String threadsQueueName;    private final String timeoutSetName;        public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name) {        this(commandExecutor, name, 60000*5);//传入60秒*5=5分钟    }        public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name, long threadWaitTime) {        super(commandExecutor, name);        this.commandExecutor = commandExecutor;        this.threadWaitTime = threadWaitTime;        threadsQueueName = prefixName("redisson_lock_queue", name);        timeoutSetName = prefixName("redisson_lock_timeout", name);    }    ...    @Override     RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {        long wait = threadWaitTime;        if (waitTime != -1) {            //将传入的指定的获取锁等待时间赋值给wait变量            wait = unit.toMillis(waitTime);        }          ...        if (command == RedisCommands.EVAL_LONG) {            return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,                //步骤一：remove stale threads，移除等待超时的线程                "while true do " +                    //获取队列中的第一个元素                    //KEYS[2]是一个用来对线程排队的队列的名字                    "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +                    "if firstThreadId2 == false then " +                        "break;" +                    "end;" +                    //获取队列中第一个元素对应的分数，也就是排第一的线程的过期时间                    //KEYS[3]是一个用来对线程排序的有序集合的名字                    "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +                    //如果排第一的线程的过期时间小于当前时间，说明该线程等待超时了都还没获取到锁，所以要移除                    //ARGV[4]是当前时间                    "if timeout