分布式锁—2.Redisson的可重入锁

大纲
1.Redisson可重入锁RedissonLock概述
2.可重入锁源码之创建RedissonClient实例
3.可重入锁源码之lua脚本加锁逻辑
4.可重入锁源码之WatchDog维持加锁逻辑
5.可重入锁源码之可重入加锁逻辑
6.可重入锁源码之锁的互斥阻塞逻辑
7.可重入锁源码之释放锁逻辑
8.可重入锁源码之获取锁超时与锁超时自动释放逻辑
9.可重入锁源码总结
 
1.Redisson可重入锁RedissonLock概述
(1)在pom.xml里引入依赖
(2)构建RedissonClient并使用Redisson
(3)Redisson可重入锁RedissonLock简单使用
 
(1)在pom.xml里引入依赖

1. <dependencies>
2.     <dependency>
3.         <groupId>org.redisson</groupId>
4.         redisson</artifactId>
5.         <version>3.16.8</version>
6.      </dependency>
7. </dependencies>

复制代码

(2)构建RedissonClient并使用Redisson
参考官网中文文档，连接上3主3从的Redis Cluster。

1. //https://github.com/redisson/redisson/wiki/目录
2. public class Application {
3.     public static void main(String[] args) throws Exception {
4.         //连接3主3从的Redis CLuster
5.         Config config = new Config();
6.         config.useClusterServers()
7.             .addNodeAddress("redis://192.168.1.110:7001")
8.             .addNodeAddress("redis://192.168.1.110:7002")
9.             .addNodeAddress("redis://192.168.1.110:7003")
10.             .addNodeAddress("redis://192.168.1.111:7001")
11.             .addNodeAddress("redis://192.168.1.111:7002")
12.             .addNodeAddress("redis://192.168.1.111:7003");
14.         //创建RedissonClient实例
15.         RedissonClient redisson = Redisson.create(config);
17.         //获取可重入锁
18.         RLock lock = redisson.getLock("myLock");
19.         lock.lock();
20.         lock.unlock();
21.       
22.         RMap<String, Object> map = redisson.getMap("myMap");
23.         map.put("foo", "bar");  
24.       
25.         map = redisson.getMap("myMap");
26.         System.out.println(map.get("foo"));   
27.     }
28. }

复制代码

(3)Redisson可重入锁RedissonLock简单使用
Redisson可重入锁RLock实现了java.util.concurrent.locks.Lock接口，同时还提供了异步(Async)、响应式(Reactive)和RxJava2标准的接口。

1. RLock lock = redisson.getLock("myLock");
2. //最常见的使用方法
3. lock.lock();

复制代码

如果设置锁的超时时间不合理，导致超时时间已到时锁还没能主动释放，但实际上锁却被Redis节点通过过期时间释放了，这会有问题。
 
为了避免这种情况，Redisson内部提供了一个用来监控锁的WatchDog。WatchDog的作用是在Redisson实例被关闭前，不断地延长锁的有效期。
 
WatchDog检查锁的默认超时时间是30秒，可通过Config.lockWatchdogTimeout来指定。
 
RLock的tryLock方法提供了leaseTime参数来指定加锁的超时时间，超过这个时间后锁便自动被释放。

1. //如果没有主动释放锁的话，10秒后将会自动释放锁
2. lock.lock(10, TimeUnit.SECONDS);
4. //加锁等待最多是100秒；加锁成功后如果没有主动释放锁的话，锁会在10秒后自动释放
5. boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
6. if (res) {
7.     try {
8.         ...
9.     } finally {
10.         lock.unlock();
11.     }
12. }

复制代码

RLock完全符合Java的Lock规范，即只有拥有锁的进程才能解锁，其他进程解锁则会抛出IllegalMonitorStateException错误。如果需要其他进程也能解锁，那么可以使用分布式信号量Semaphore。
 
2.可重入锁源码之创建RedissonClient实例
(1)初始化与Redis的连接管理器ConnectionManager
(2)初始化Redis的命令执行器CommandExecutor
 
使用Redisson.create()方法可以根据配置创建一个RedissonClient实例，因为Redisson类会实现RedissonClient接口，而创建RedissonClient实例的主要工作其实就是：
一.初始化与Redis的连接管理器ConnectionManager
二.初始化Redis的命令执行器CommandExecutor
 
(1)初始化与Redis的连接管理器ConnectionManager
Redis的配置类Config会被封装在连接管理器ConnectionManager中，后续可以通过连接管理器ConnectionManager获取Redis的配置类Config。

1. public class Application {
2.     public static void main(String[] args) throws Exception {
3.         Config config = new Config();
4.         config.useClusterServers().addNodeAddress("redis://192.168.1.110:7001");
5.         //创建RedissonClient实例
6.         RedissonClient redisson = Redisson.create(config);
7.         ...
8.     }
9. }
11. //创建RedissonClient实例的源码
12. public class Redisson implements RedissonClient {
13.     protected final Config config;//Redis配置类
14.     protected final ConnectionManager connectionManager;//Redis的连接管理器
15.     protected final CommandAsyncExecutor commandExecutor;//Redis的命令执行器
16.   
17.     ...
18.     public static RedissonClient create(Config config) {
19.         return new Redisson(config);
20.     }
21.    
22.     protected Redisson(Config config) {
23.         this.config = config;
24.         Config configCopy = new Config(config);
25.         //根据Redis配置类Config实例创建和Redis的连接管理器
26.         connectionManager = ConfigSupport.createConnectionManager(configCopy);
27.         RedissonObjectBuilder objectBuilder = null;
28.         if (config.isReferenceEnabled()) {
29.             objectBuilder = new RedissonObjectBuilder(this);
30.         }
31.         //创建Redis的命令执行器
32.         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
33.         evictionScheduler = new EvictionScheduler(commandExecutor);
34.         writeBehindService = new WriteBehindService(commandExecutor);
35.     }
36.     ...
37. }
39. public class ConfigSupport {
40.     ...
41.     //创建Redis的连接管理器
42.     public static ConnectionManager createConnectionManager(Config configCopy) {
43.         //生成UUID
44.         UUID id = UUID.randomUUID();
45.         ...
46.         if (configCopy.getClusterServersConfig() != null) {
47.             validate(configCopy.getClusterServersConfig());
48.             //返回ClusterConnectionManager实例
49.             return new ClusterConnectionManager(configCopy.getClusterServersConfig(), configCopy, id);
50.         }
51.         ...
52.     }
53.     ...
54. }
56. public class ClusterConnectionManager extends MasterSlaveConnectionManager {
57.     public ClusterConnectionManager(ClusterServersConfig cfg, Config config, UUID id) {
58.         super(config, id);
59.         ...
60.         this.natMapper = cfg.getNatMapper();
61.         //将Redis的配置类Config封装在ConnectionManager中
62.         this.config = create(cfg);
63.         initTimer(this.config);
64.    
65.         Throwable lastException = null;
66.         List<String> failedMasters = new ArrayList<String>();
67.         for (String address : cfg.getNodeAddresses()) {
68.             RedisURI addr = new RedisURI(address);
69.             //异步连接Redis节点
70.             CompletionStage<RedisConnection> connectionFuture = connectToNode(cfg, addr, addr.getHost());
71.             ...
72.             //通过connectionFuture阻塞获取建立好的连接
73.             RedisConnection connection = connectionFuture.toCompletableFuture().join();
74.             ...
75.             List<ClusterNodeInfo> nodes = connection.sync(clusterNodesCommand);
76.             ...
77.             CompletableFuture<Collection<ClusterPartition>> partitionsFuture = parsePartitions(nodes);
78.             Collection<ClusterPartition> partitions = partitionsFuture.join();
79.             List<CompletableFuture<Void>> masterFutures = new ArrayList<>();
80.             for (ClusterPartition partition : partitions) {
81.                 if (partition.isMasterFail()) {
82.                     failedMasters.add(partition.getMasterAddress().toString());
83.                     continue;
84.                 }
85.                 if (partition.getMasterAddress() == null) {
86.                     throw new IllegalStateException("Master node: " + partition.getNodeId() + " doesn't have address.");
87.                 }
88.                 CompletableFuture<Void> masterFuture = addMasterEntry(partition, cfg);
89.                 masterFutures.add(masterFuture);
90.             }
91.             CompletableFuture<Void> masterFuture = CompletableFuture.allOf(masterFutures.toArray(new CompletableFuture[0]));
92.             masterFuture.join();
93.             ...
94.         }
95.         ...
96.     }
97.     ...
98. }
100. public class MasterSlaveConnectionManager implements ConnectionManager {
101.     protected final String id;//初始化时为UUID
102.     private final Map<RedisURI, RedisConnection> nodeConnections = new ConcurrentHashMap<>();
103.     ...
104.     protected MasterSlaveConnectionManager(Config cfg, UUID id) {
105.         this.id = id.toString();//传入的是UUID
106.         this.cfg = cfg;
107.         ...
108.     }
109.    
110.     protected final CompletionStage<RedisConnection> connectToNode(NodeType type, BaseConfig<?> cfg, RedisURI addr, String sslHostname) {
111.         RedisConnection conn = nodeConnections.get(addr);
112.         if (conn != null) {
113.             if (!conn.isActive()) {
114.                 closeNodeConnection(conn);
115.             } else {
116.                 return CompletableFuture.completedFuture(conn);
117.             }
118.         }
119.         //创建Redis客户端连接实例
120.         RedisClient client = createClient(type, addr, cfg.getConnectTimeout(), cfg.getTimeout(), sslHostname);
121.         //向Redis服务端发起异步连接请求，这个future会层层往外返回
122.         CompletionStage<RedisConnection> future = client.connectAsync();
123.         return future.thenCompose(connection -> {
124.             if (connection.isActive()) {
125.                 if (!addr.isIP()) {
126.                     RedisURI address = new RedisURI(addr.getScheme() + "://" + connection.getRedisClient().getAddr().getAddress().getHostAddress() + ":" + connection.getRedisClient().getAddr().getPort());
127.                     nodeConnections.put(address, connection);
128.                 }
129.                 nodeConnections.put(addr, connection);
130.                 return CompletableFuture.completedFuture(connection);
131.             } else {
132.                 connection.closeAsync();
133.                 CompletableFuture<RedisConnection> f = new CompletableFuture<>();
134.                 f.completeExceptionally(new RedisException("Connection to " + connection.getRedisClient().getAddr() + " is not active!"));
135.                 return f;
136.             }
137.         });
138.     }
139.    
140.     //创建Redis客户端连接实例
141.     @Override
142.     public RedisClient createClient(NodeType type, RedisURI address, int timeout, int commandTimeout, String sslHostname) {
143.         RedisClientConfig redisConfig = createRedisConfig(type, address, timeout, commandTimeout, sslHostname);
144.         return RedisClient.create(redisConfig);
145.     }
146.     ...
147. }
149. //Redisson主要使用Netty去和Redis服务端建立连接
150. public final class RedisClient {
151.     private final Bootstrap bootstrap;
152.     private final Bootstrap pubSubBootstrap;
153.     ...
154.     public static RedisClient create(RedisClientConfig config) {
155.         return new RedisClient(config);
156.     }
157.    
158.     private RedisClient(RedisClientConfig config) {
159.         ...
160.         bootstrap = createBootstrap(copy, Type.PLAIN);
161.         pubSubBootstrap = createBootstrap(copy, Type.PUBSUB);
162.         this.commandTimeout = copy.getCommandTimeout();
163.     }
164.    
165.     private Bootstrap createBootstrap(RedisClientConfig config, Type type) {
166.         Bootstrap bootstrap = new Bootstrap()
167.             .resolver(config.getResolverGroup())
168.             .channel(config.getSocketChannelClass())
169.             .group(config.getGroup());
170.         bootstrap.handler(new RedisChannelInitializer(bootstrap, config, this, channels, type));
171.         bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout());
172.         bootstrap.option(ChannelOption.SO_KEEPALIVE, config.isKeepAlive());
173.         bootstrap.option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay());
174.         config.getNettyHook().afterBoostrapInitialization(bootstrap);
175.         return bootstrap;
176.     }
177.    
178.     //向Redis服务端发起异步连接请求
179.     public RFuture<RedisConnection> connectAsync() {
180.         CompletableFuture<InetSocketAddress> addrFuture = resolveAddr();
181.         CompletableFuture<RedisConnection> f = addrFuture.thenCompose(res -> {
182.             CompletableFuture<RedisConnection> r = new CompletableFuture<>();
183.             //Netty的Bootstrap发起连接
184.             ChannelFuture channelFuture = bootstrap.connect(res);
185.             channelFuture.addListener(new ChannelFutureListener() {
186.                 @Override
187.                 public void operationComplete(final ChannelFuture future) throws Exception {
188.                     if (bootstrap.config().group().isShuttingDown()) {
189.                         IllegalStateException cause = new IllegalStateException("RedisClient is shutdown");
190.                         r.completeExceptionally(cause);
191.                         return;
192.                     }
193.                     if (future.isSuccess()) {
194.                         RedisConnection c = RedisConnection.getFrom(future.channel());
195.                         c.getConnectionPromise().whenComplete((res, e) -> {
196.                             bootstrap.config().group().execute(new Runnable() {
197.                                 @Override
198.                                 public void run() {
199.                                     if (e == null) {
200.                                         if (!r.complete(c)) {
201.                                             c.closeAsync();
202.                                         }
203.                                     } else {
204.                                         r.completeExceptionally(e);
205.                                         c.closeAsync();
206.                                     }
207.                                 }
208.                             });
209.                         });
210.                     } else {
211.                         bootstrap.config().group().execute(new Runnable() {
212.                             public void run() {
213.                                 r.completeExceptionally(future.cause());
214.                             }
215.                         });
216.                     }
217.                 }
218.             });
219.             return r;
220.         });
221.         return new CompletableFutureWrapper<>(f);
222.     }
223.     ...
224. }

复制代码

(2)初始化Redis的命令执行器CommandExecutor
首先，CommandSyncService继承自CommandAsyncService类。
 
而CommandAsyncService类实现了CommandExecutor接口。
 
然后，ConnectionManager连接管理器会封装在命令执行器CommandExecutor中。
 
所以，通过CommandExecutor命令执行器可以获取连接管理器ConnectionManager。

1. //Redis命令的同步执行器CommandSyncService
2. public class CommandSyncService extends CommandAsyncService implements CommandExecutor {
3.     //初始化CommandExecutor
4.     public CommandSyncService(ConnectionManager connectionManager, RedissonObjectBuilder objectBuilder) {
5.         super(connectionManager, objectBuilder, RedissonObjectBuilder.ReferenceType.DEFAULT);
6.     }
7.    
8.     public <T, R> R read(String key, RedisCommand<T> command, Object... params) {
9.         return read(key, connectionManager.getCodec(), command, params);
10.     }
11.    
12.     public <T, R> R read(String key, Codec codec, RedisCommand<T> command, Object... params) {
13.         RFuture<R> res = readAsync(key, codec, command, params);
14.         return get(res);
15.     }
16.    
17.     public <T, R> R evalRead(String key, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object... params) {
18.         return evalRead(key, connectionManager.getCodec(), evalCommandType, script, keys, params);
19.     }
20.    
21.     public <T, R> R evalRead(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object... params) {
22.         RFuture<R> res = evalReadAsync(key, codec, evalCommandType, script, keys, params);
23.         return get(res);
24.     }
25.    
26.     public <T, R> R evalWrite(String key, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object... params) {
27.         return evalWrite(key, connectionManager.getCodec(), evalCommandType, script, keys, params);
28.     }
29.    
30.     public <T, R> R evalWrite(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object... params) {
31.         RFuture<R> res = evalWriteAsync(key, codec, evalCommandType, script, keys, params);
32.         return get(res);
33.     }
34. }
36. //Redis命令的异步执行器CommandAsyncService
37. public class CommandAsyncService implements CommandAsyncExecutor {
38.     //Redis连接管理器
39.     final ConnectionManager connectionManager;
40.     final RedissonObjectBuilder objectBuilder;
41.     final RedissonObjectBuilder.ReferenceType referenceType;
43.     public CommandAsyncService(ConnectionManager connectionManager, RedissonObjectBuilder objectBuilder, RedissonObjectBuilder.ReferenceType referenceType) {
44.         this.connectionManager = connectionManager;
45.         this.objectBuilder = objectBuilder;
46.         this.referenceType = referenceType;
47.     }
48.    
49.     @Override
50.     public <V> V getNow(CompletableFuture<V> future) {
51.         try {
52.             return future.getNow(null);
53.         } catch (Exception e) {
54.             return null;
55.         }
56.     }
57.    
58.     @Override
59.     public <T, R> R read(String key, Codec codec, RedisCommand<T> command, Object... params) {
60.         RFuture<R> res = readAsync(key, codec, command, params);
61.         return get(res);
62.     }
63.    
64.     @Override
65.     public <T, R> RFuture<R> readAsync(String key, Codec codec, RedisCommand<T> command, Object... params) {
66.         NodeSource source = getNodeSource(key);
67.         return async(true, source, codec, command, params, false, false);
68.     }
69.    
70.     private NodeSource getNodeSource(String key) {
71.         int slot = connectionManager.calcSlot(key);
72.         return new NodeSource(slot);
73.     }
74.    
75.     public <V, R> RFuture<R> async(boolean readOnlyMode, NodeSource source, Codec codec, RedisCommand<V> command, Object[] params, boolean ignoreRedirect, boolean noRetry) {
76.         CompletableFuture<R> mainPromise = createPromise();
77.         RedisExecutor<V, R> executor = new RedisExecutor<>(readOnlyMode, source, codec, command, params, mainPromise, ignoreRedirect, connectionManager, objectBuilder, referenceType, noRetry);
78.         executor.execute();
79.         return new CompletableFutureWrapper<>(mainPromise);
80.     }
81.    
82.     @Override
83.     public <V> V get(RFuture<V> future) {
84.         if (Thread.currentThread().getName().startsWith("redisson-netty")) {
85.             throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");
86.         }
87.         try {
88.             return future.toCompletableFuture().get();
89.         } catch (InterruptedException e) {
90.             future.cancel(true);
91.             Thread.currentThread().interrupt();
92.             throw new RedisException(e);
93.         } catch (ExecutionException e) {
94.             throw convertException(e);
95.         }
96.     }
97.     ...
98. }

复制代码

 
3.可重入锁源码之lua脚本加锁逻辑
(1)通过Redisson.getLock()方法获取一个RedissonLock实例
(2)加锁时的执行流程
(3)加锁时执行的lua脚本
(4)执行加锁lua脚本的命令执行器逻辑
(5)如何根据slot值获取对应的节点
 
(1)通过Redisson.getLock()方法获取一个RedissonLock实例
在Redisson.getLock()方法中，会传入命令执行器CommandExecutor来创建一个RedissonLock实例，而命令执行器CommandExecutor是在执行Redisson.create()方法时初始化好的，所以命令执行器CommandExecutor会被封装在RedissonLock实例中。
 
因此，通过RedissonLock实例可以获取一个命令执行器CommandExecutor，通过命令执行器CommandExecutor可获取连接管理器ConnectionManager，通过连接管理器ConnectionManager可获取Redis的配置信息类Config，通过Redis的配置信息类Config可以获取各种配置信息。
 
RedissonLock类继承自实现了RLock接口的RedissonBaseLock类。在RedissonLock的构造方法里面，有个internalLockLeaseTime变量，这个internalLockLeaseTime变量与WatchDog看门狗有关系。interlnalLockLeaseTime的默认值是30000毫秒，即30秒；

1. public class Application {
2.     public static void main(String[] args) throws Exception {
3.         Config config = new Config();
4.         config.useClusterServers().addNodeAddress("redis://192.168.1.110:7001");
5.         //创建RedissonClient实例
6.         RedissonClient redisson = Redisson.create(config);
7.         //获取可重入锁
8.         RLock lock = redisson.getLock("myLock");
9.         lock.lock();
10.         ...
11.     }
12. }
14. //创建Redisson实例
15. public class Redisson implements RedissonClient {
16.     protected final Config config;//Redis配置类
17.     protected final ConnectionManager connectionManager;//Redis的连接管理器
18.     protected final CommandAsyncExecutor commandExecutor;//Redis的命令执行器
19.     ...
20.    
21.     public static RedissonClient create(Config config) {
22.         return new Redisson(config);
23.     }
24.    
25.     protected Redisson(Config config) {
26.         ...
27.         //根据Redis配置类Config实例创建和Redis的连接管理器
28.         connectionManager = ConfigSupport.createConnectionManager(configCopy);
29.         //创建Redis的命令执行器
30.         commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
31.         ...
32.     }
33.     ...
34.     @Override
35.     public RLock getLock(String name) {
36.         return new RedissonLock(commandExecutor, name);
37.     }
38.     ...
39. }
41. //创建RedissonLock实例
42. //通过RedissonLock实例可以获取一个命令执行器CommandExecutor；
43. public class RedissonLock extends RedissonBaseLock {
44.     protected long internalLockLeaseTime;
45.     protected final LockPubSub pubSub;
46.     final CommandAsyncExecutor commandExecutor;
47.    
48.     public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
49.         super(commandExecutor, name);
50.         this.commandExecutor = commandExecutor;
51.         //与WatchDog有关的internalLockLeaseTime
52.         //通过命令执行器CommandExecutor可以获取连接管理器ConnectionManager
53.         //通过连接管理器ConnectionManager可以获取Redis的配置信息类Config
54.         //通过Redis的配置信息类Config可以获取lockWatchdogTimeout超时时间
55.         this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
56.         this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
57.     }
58.     ...
59. }
61. //创建Redis的命令执行器
62. //通过命令执行器CommandExecutor可以获取连接管理器ConnectionManager
63. public class CommandAsyncService implements CommandAsyncExecutor {
64.     final ConnectionManager connectionManager;
65.     ...
66.     public CommandAsyncService(ConnectionManager connectionManager, RedissonObjectBuilder objectBuilder, RedissonObjectBuilder.ReferenceType referenceType) {
67.         this.connectionManager = connectionManager;
68.         this.objectBuilder = objectBuilder;
69.         this.referenceType = referenceType;
70.     }
71.    
72.     @Override
73.     public ConnectionManager getConnectionManager() {
74.         return connectionManager;
75.     }
76.     ...
77. }
79. //创建Redis的连接管理器
80. //通过连接管理器ConnectionManager可以获取Redis的配置信息类Config
81. public class ClusterConnectionManager extends MasterSlaveConnectionManager {
82.     ...
83.     public ClusterConnectionManager(ClusterServersConfig cfg, Config config, UUID id) {
84.         super(config, id);
85.         ...
86.     }
87.     ...
88. }
90. //创建Redis的连接管理器
91. //通过连接管理器ConnectionManager可以获取Redis的配置信息类Config
92. public class MasterSlaveConnectionManager implements ConnectionManager {
93.     private final Config cfg;
94.     protected final String id;//初始化时为UUID
95.     ...
96.     protected MasterSlaveConnectionManager(Config cfg, UUID id) {
97.         this.id = id.toString();//传入的是UUID
98.         this.cfg = cfg;
99.         ...
100.     }
101.    
102.     @Override
103.     public Config getCfg() {
104.         return cfg;
105.     }
106.     ...
107. }
109. //配置信息类Config中的lockWatchdogTimeout变量初始化为30秒，该变量与WatchDog有关
110. public class Config {
111.     private long lockWatchdogTimeout = 30 * 1000;
112.     ...
113.     //This parameter is only used if lock has been acquired without leaseTimeout parameter definition.
114.     //Lock expires after "lockWatchdogTimeout" if watchdog didn't extend it to next "lockWatchdogTimeout" time interval.
115.     //This prevents against infinity locked locks due to Redisson client crush or any other reason when lock can't be released in proper way.
116.     //Default is 30000 milliseconds
117.     public Config setLockWatchdogTimeout(long lockWatchdogTimeout) {
118.         this.lockWatchdogTimeout = lockWatchdogTimeout;
119.         return this;
120.     }
121.    
122.     public long getLockWatchdogTimeout() {
123.         return lockWatchdogTimeout;
124.     }
125. }

复制代码

默认情况下，调用RedissonLock.lock()方法加锁时，传入的leaseTime为-1。此时锁的超时时间会设为lockWatchdogTimeout默认的30秒，从而避免出现死锁的情况。

1. public class RedissonLock extends RedissonBaseLock {
2.     ...
3.     //加锁
4.     @Override
5.     public void lock() {
6.         try {
7.             lock(-1, null, false);
8.         } catch (InterruptedException e) {
9.             throw new IllegalStateException();
10.         }
11.     }
12.    
13.     private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
14.         long threadId = Thread.currentThread().getId();
15.         Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
16.         ...
17.     }
18.    
19.     //解锁
20.     @Override
21.     public void unlock() {
22.         try {
23.             get(unlockAsync(Thread.currentThread().getId()));
24.         } catch (RedisException e) {
25.             if (e.getCause() instanceof IllegalMonitorStateException) {
26.                 throw (IllegalMonitorStateException) e.getCause();
27.             } else {
28.                 throw e;
29.             }
30.         }
31.     }
32.     ...
33. }

复制代码

(2)加锁时的执行流程
首先会调用RedissonLock的tryAcquire()方法处理异步RFuture相关，然后调用RedissonLock的tryAcquireAsync()方法对执行脚本的结果进行处理，接着调用RedissonLock.tryLockInnerAsync方法执行加锁的lua脚本。

1. public class RedissonLock extends RedissonBaseLock {
2.     protected long internalLockLeaseTime;
3.     protected final LockPubSub pubSub;
4.     final CommandAsyncExecutor commandExecutor;
5.    
6.     public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
7.         super(commandExecutor, name);
8.         this.commandExecutor = commandExecutor;
9.         //与WatchDog有关的internalLockLeaseTime
10.         //通过命令执行器CommandExecutor可以获取连接管理器ConnectionManager
11.         //通过连接管理器ConnectionManager可以获取Redis的配置信息类Config
12.         //通过Redis的配置信息类Config可以获取lockWatchdogTimeout超时时间
13.         this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
14.         this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
15.     }
16.     ...
17.     //加锁
18.     @Override
19.     public void lock() {
20.         try {
21.             lock(-1, null, false);
22.         } catch (InterruptedException e) {
23.             throw new IllegalStateException();
24.         }
25.     }
26.    
27.     private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
28.         //线程ID，用来生成设置Hash的值
29.         long threadId = Thread.currentThread().getId();
30.         //尝试加锁，此时执行RedissonLock.lock()方法默认传入的leaseTime=-1
31.         Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
32.         //ttl为null说明加锁成功
33.         if (ttl == null) {
34.             return;
35.         }
36.         //加锁失败时的处理
37.         CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
38.         if (interruptibly) {
39.             commandExecutor.syncSubscriptionInterrupted(future);
40.         } else {
41.             commandExecutor.syncSubscription(future);
42.         }
43.         try {
44.             while (true) {
45.                 ttl = tryAcquire(-1, leaseTime, unit, threadId);
46.                 // lock acquired
47.                 if (ttl == null) {
48.                     break;
49.                 }
50.                 // waiting for message
51.                 if (ttl >= 0) {
52.                     try {
53.                         commandExecutor.getNow(future).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
54.                     } catch (InterruptedException e) {
55.                         if (interruptibly) {
56.                             throw e;
57.                         }
58.                         commandExecutor.getNow(future).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
59.                     }
60.                 } else {
61.                     if (interruptibly) {
62.                         commandExecutor.getNow(future).getLatch().acquire();
63.                     } else {
64.                         commandExecutor.getNow(future).getLatch().acquireUninterruptibly();
65.                     }
66.                 }
67.             }
68.         } finally {
69.             unsubscribe(commandExecutor.getNow(future), threadId);
70.         }
71.     }
72.     ...
73.     private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
74.         //默认下waitTime和leaseTime都是-1，下面调用的get方法是来自于RedissonObject的get()方法
75.         //可以理解为异步转同步：将异步的tryAcquireAsync通过get转同步
76.         return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
77.     }
78.    
79.     private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
80.         RFuture<Long> ttlRemainingFuture;
81.         if (leaseTime != -1) {
82.             ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
83.         } else {
84.             //默认情况下，由于leaseTime=-1，所以会使用初始化RedissonLock实例时的internalLockLeaseTime
85.             //internalLockLeaseTime的默认值就是lockWatchdogTimeout的默认值，30秒
86.             ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
87.         }
88.         CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
89.             //加锁返回的ttlRemaining为null表示加锁成功
90.             if (ttlRemaining == null) {
91.                 if (leaseTime != -1) {
92.                     internalLockLeaseTime = unit.toMillis(leaseTime);
93.                 } else {
94.                     scheduleExpirationRenewal(threadId);
95.                 }
96.             }
97.             return ttlRemaining;
98.         });
99.         return new CompletableFutureWrapper<>(f);
100.     }
101.    
102.     //默认情况下，外部传入的leaseTime=-1时，会取lockWatchdogTimeout的默认值=30秒
103.     <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
104.         return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
105.             "if (redis.call('exists', KEYS[1]) == 0) then " +
106.                 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
107.                 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
108.                 "return nil; " +
109.             "end; " +
110.             "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
111.                 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
112.                 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
113.                 "return nil; " +
114.             "end; " +
115.             "return redis.call('pttl', KEYS[1]);",
116.             Collections.singletonList(getRawName()),//锁的名字：KEYS[1]
117.             unit.toMillis(leaseTime),//过期时间：ARGV[1]，默认时为30秒
118.             getLockName(threadId)//ARGV[2]，值为UUID + 线程ID
119.         );
120.     }
121.     ...
122. }
124. public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {
125.     final String id;
126.     final String entryName;
127.     final CommandAsyncExecutor commandExecutor;
128.    
129.     public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
130.         super(commandExecutor, name);
131.         this.commandExecutor = commandExecutor;
132.         this.id = commandExecutor.getConnectionManager().getId();
133.         this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
134.         this.entryName = id + ":" + name;
135.     }
136.    
137.     protected String getLockName(long threadId) {
138.         return id + ":" + threadId;
139.     }
140.     ...
141. }
143. abstract class RedissonExpirable extends RedissonObject implements RExpirable {
144.     ...
145.     RedissonExpirable(CommandAsyncExecutor connectionManager, String name) {
146.         super(connectionManager, name);
147.     }
148.     ...
149. }
151. public abstract class RedissonObject implements RObject {
152.     protected final CommandAsyncExecutor commandExecutor;
153.     protected String name;
154.     protected final Codec codec;
155.    
156.     public RedissonObject(Codec codec, CommandAsyncExecutor commandExecutor, String name) {
157.         this.codec = codec;
158.         this.commandExecutor = commandExecutor;
159.         if (name == null) {
160.             throw new NullPointerException("name can't be null");
161.         }
162.         setName(name);
163.     }
164.     ...
165.     protected final <V> V get(RFuture<V> future) {
166.         //下面会调用CommandAsyncService.get()方法
167.         return commandExecutor.get(future);
168.     }
169.     ...
170. }
172. public class CommandAsyncService implements CommandAsyncExecutor {
173.     ...
174.     @Override
175.     public <V> V get(RFuture<V> future) {
176.         if (Thread.currentThread().getName().startsWith("redisson-netty")) {
177.             throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");
178.         }
179.         try {
180.             return future.toCompletableFuture().get();
181.         } catch (InterruptedException e) {
182.             future.cancel(true);
183.             Thread.currentThread().interrupt();
184.             throw new RedisException(e);
185.         } catch (ExecutionException e) {
186.             throw convertException(e);
187.         }
188.     }
189.     ...
190. }

复制代码

(3)加锁时执行的lua脚本

1. public class RedissonLock extends RedissonBaseLock {
2.     //默认情况下，外部传入的leaseTime=-1时，会取lockWatchdogTimeout的默认值=30秒
3.     <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
4.         return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
5.             "if (redis.call('exists', KEYS[1]) == 0) then " +
6.                 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
7.                 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
8.                 "return nil; " +
9.             "end; " +
10.             "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
11.                 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
12.                 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
13.                 "return nil; " +
14.             "end; " +
15.             "return redis.call('pttl', KEYS[1]);",
16.             Collections.singletonList(getRawName()),//锁的名字：KEYS[1]，比如"myLock"
17.             unit.toMillis(leaseTime),//过期时间：ARGV[1]，默认时为30秒
18.             getLockName(threadId)//ARGV[2]，值为UUID + 线程ID
19.         );
20.     }
21.     ...
22. }

复制代码

首先执行Redis的exists命令，判断key为锁名的Hash值是否不存在，也就是判断key为锁名myLock的Hash值是否存在。
 
一.如果key为锁名的Hash值不存在，那么就进行如下加锁处理
首先通过Redis的hset命令设置一个key为锁名的Hash值。该Hash值的key为锁名，value是一个映射。也就是在value值中会有一个field为UUID + 线程ID，value为1的映射。比如：hset myLock UUID:ThreadID 1，lua脚本中的ARGV[2]就是由UUID + 线程ID组成的唯一值。
 
然后通过Redis的pexpire命令设置key为锁名的Hash值的过期时间，也就是设置key为锁名的Hash值的过期时间为30秒。比如：pexpire myLock 30000。所以默认情况下，myLock这个锁在30秒后就会自动过期。
 
二.如果key为锁名的Hash值存在，那么就执行如下判断处理
首先通过Redis的hexists命令判断在key为锁名的Hash值里，field为UUID + 线程ID的映射是否已经存在。
 
如果在key为锁名的Hash值里，field为UUID + 线程ID的映射存在，那么就通过Redis的hincrby命令，对field为UUID + 线程ID的value值进行递增1。比如：hincrby myLock UUID:ThreadID 1。也就是在key为myLock的Hash值里，把field为UUID:ThreadID的value值从1累加到2，发生这种情况的时候往往就是当前线程对锁进行了重入。接着执行：pexpire myLock 30000，再次将myLock的有效期设置为30秒。
 
如果在key为锁名的Hash值里，field为UUID + 线程ID的映射不存在，发生这种情况的时候往往就是其他线程获取不到这把锁而产生互斥。那么就通过Redis的pttl命令，返回key为锁名的Hash值的剩余存活时间，因为不同线程的ARGV[2]是不一样的，ARGV[2] = UUID + 线程ID。
 
(4)执行加锁lua脚本的命令执行器逻辑
在RedissonLock的tryLockInnerAsync()方法中，会通过RedissonBaseLock的evalWriteAsync()方法执行lua脚本，即通过CommandAsyncService的evalWriteAsync()方法执行lua脚本。
 
在CommandAsyncService的evalWriteAsync()方法中，首先会执行CommandAsyncService的getNodeSource()方法获取对应的节点。然后执行CommandAsyncService的evalAsync()方法来执行lua脚本。
 
在CommandAsyncService的getNodeSource()方法中，会根据key进行CRC16运算，然后再对16384取模，计算出key的slot值。然后根据这个slot值创建一个NodeSource实例进行返回。
 
在CommandAsyncService的evalAsync()方法中，会将获得的NodeSource实例封装到Redis执行器RedisExecutor里。然后执行RedisExecutor，实现将脚本请求发送给对应的Redis节点处理。
[code]public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {    //从外部传入的：在创建实现了RedissonClient的Redisson实例时，初始化的命令执行器CommandExecutor    final CommandAsyncExecutor commandExecutor;    public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {        super(commandExecutor, name);        this.commandExecutor = commandExecutor;        this.id = commandExecutor.getConnectionManager().getId();        this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();        this.entryName = id + ":" + name;    }    ...    protected  RFuture evalWriteAsync(String key, Codec codec, RedisCommand evalCommandType, String script, List keys, Object... params) {        //获取可用的节点，并继续封装一个命令执行器CommandBatchService        MasterSlaveEntry entry = commandExecutor.getConnectionManager().getEntry(getRawName());        int availableSlaves = entry.getAvailableSlaves();        CommandBatchService executorService = createCommandBatchService(availableSlaves);        //通过CommandAsyncService.evalWriteAsync方法执行lua脚本        RFuture result = executorService.evalWriteAsync(key, codec, evalCommandType, script, keys, params);        if (commandExecutor instanceof CommandBatchService) {            return result;        }        //异步执行然后获取结果        RFuture