redis 分布式重入锁

---

媒介

通过前篇文章 redis 分布式锁实现
我们发现简朴做一把分布式锁没啥标题，但是针对以往的锁来说，还存在一下两点必要思量。

• 1.一个线程如果多次重复拿锁，该怎样实现重入
  
• 2.由于防止死锁设置了逾期时间，那么如果锁的时间到期了，业务还没有实行完毕，导致新的业务进来造成的并发标题如那边理
  

---

一、分布式重入锁

1、单机重入锁

在单机锁期间，必要支持重入告急是为了制止在单线程环境下大概出现的死锁标题，同时简化编程模子，提升代码的机动性与可重用性。

• 制止死锁：如果一个线程在持有锁的环境下，再次实行获取同一把锁，非重入锁会导致该线程不停等候自己开释锁，从而引发死锁标题。重入锁答应同一个线程多次获取同一把锁，如许就可以制止这种环境的发生。
  
• 简化递归调用场景：在递归方法中，方法会多次调用自己，而每次调用都必要通过同一个锁掩护共享资源。重入锁可以或许确保这些调用不会由于锁的重复获取而出现壅闭环境。
  
• 支持锁的调用链：在面向对象编程中，一个持有锁的方法大概会调用对象中同样必要持有锁的其他方法。重入锁包管这些方法可以顺遂实行而不会由于锁的竞争而壅闭。
  
• 加强机动性：重入锁数据布局更复杂，可以纪录获取锁的次数。这使得锁可以机动用在较复杂的同步场景中。
  综上所述，重入锁通过答应同一线程多次获取同一把锁，制止了很多潜伏的同步标题，使得同步代码的编写变得更加简朴和可靠。
  

比方synchronized、ReentrantLock
2、redis重入锁

参考重入锁的筹划头脑，我们在实现redis重入锁，应该要遵照一下原则

• 互斥条件：实现锁的须要条件，标记是否有线程已占用，差别线程不能重复占用
  
• 线程信息：纪录线程信息，来判定加锁的是不是同一个线程
  
• 重入次数：纪录重入次数，再开释锁的时间，淘汰相应的次数
  

二、redisson实现重入锁

1、 添加依赖

1. <dependency>
2.     <groupId>org.redisson</groupId>
3.     <artifactId>redisson-spring-boot-starter</artifactId>
4.     <version>3.16.3</version> <!-- 根据需要选择合适的版本 -->
5. </dependency>

复制代码

2、 设置 Redisson 客户端

平凡模式

1. import org.redisson.Redisson;
2. import org.redisson.api.RedissonClient;
3. import org.redisson.config.Config;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. @Configuration
8. public class RedissonConfig {
10.     @Bean
11.     public RedissonClient redissonClient() {
12.         Config config = new Config();
13.         config.useSingleServer()
14.                 .setAddress("redis://127.0.0.1:6379");
15.         return Redisson.create(config);
16.     }
17. }

复制代码

集群模式

1. @Bean
2. public RedissonClient redissonClient() {
3.     Config config = new Config();
4.     config.useClusterServers()
5.         .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001");
6.     return Redisson.create(config);
7. }

复制代码

哨兵模式

1. @Bean
2. public RedissonClient redissonClient() {
3.     Config config = new Config();
4.     config.useSentinelServers()
5.             .setMasterName("masterName")
6.             .addSentinelAddress("redis://127.0.0.1:26379", "redis://127.0.0.1:26380");
7.     return Redisson.create(config);
8. }

复制代码

3、 使用 Redisson 实现重入锁

1. import org.redisson.api.RLock;
2. import org.redisson.api.RedissonClient;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.stereotype.Service;
6. @Service
7. public class MyService {
9.     @Autowired
10.     private RedissonClient redissonClient;
12.     private String lock = "myLock";
14.     public void outerMethod() {
15.         RLock lock = redissonClient.getLock(lock);
16.         lock.lock();
17.         try {
18.             System.out.println("Outer method acquired lock");
19.             innerMethod();
20.         } finally {
21.             lock.unlock();
22.             System.out.println("Outer method released lock");
23.         }
24.     }
26.     private void innerMethod() {
27.         RLock lock = redissonClient.getLock(lock);
28.         lock.lock();
29.         try {
30.             System.out.println("Inner method acquired lock");
31.         } finally {
32.             lock.unlock();
33.             System.out.println("Inner method released lock");
34.         }
35.     }
36. }

复制代码

4、 验证

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.web.bind.annotation.GetMapping;
3. import org.springframework.web.bind.annotation.RestController;
5. @RestController
6. public class MyController {
8.     @Autowired
9.     private MyService myService;
11.     @GetMapping("/test-lock")
12.     public String testLock() {
13.         myService.outerMethod();
14.         return "Lock tested successfully";
15.     }
16. }

复制代码

5、运行项目

启动 Spring Boot 应用，并访问 http://localhost:8080/test-lock 以测试多次重入锁的实现。你应该可以或许在控制台上看到如下输出，表明锁多次重入的准确实行：

1. Outer method acquired lock
2. Inner method acquired lock
3. Inner method released lock
4. Outer method released lock

复制代码

三、redisson分布式锁分析

1、获取锁对象

1. RLock lock = redissonClient.getLock(lock);
3. public RLock getLock(String name) {
4.   return new RedissonLock(this.connectionManager.getCommandExecutor(), name, this.id);
5. }
7. public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
8.   super(commandExecutor, name);
9.   this.commandExecutor = commandExecutor;
10.   this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
11.   this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
12. }
14. public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
15.   super(commandExecutor, name);
16.   this.commandExecutor = commandExecutor;
17.   this.id = commandExecutor.getConnectionManager().getId();
18.   // 默认锁释放时间 30s
19.   this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
20.   this.entryName = id + ":" + name;
21. }
23. public RedissonObject(Codec codec, CommandAsyncExecutor commandExecutor, String name) {
24.   this.codec = codec;
25.   this.commandExecutor = commandExecutor;
26.   if (name == null) {
27.     throw new NullPointerException("name can't be null");
28.   }
30.   setName(name);
31. }

复制代码

2、 加锁

org.redisson.RedissonLock#tryLock

1.     @Override
2.     public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
3.         // 等待时间转成MS
4.         long time = unit.toMillis(waitTime);
5.         long current = System.currentTimeMillis();
6.         // 当前线程
7.         long threadId = Thread.currentThread().getId();
8.         // 尝试获取锁 返回空标识回去锁
9.         Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
10.         // lock acquired
11.         if (ttl == null) {
12.             return true;
13.         }
14.         
15.         time -= System.currentTimeMillis() - current;
16.         // 超时，拿锁失败 返回false
17.         if (time <= 0) {
18.             acquireFailed(waitTime, unit, threadId);
19.             return false;
20.         }
21.         
22.         current = System.currentTimeMillis();
23.         // 订阅解锁消息，见org.redisson.pubsub.LockPubSub#onMessage
24.         // 订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：
25.         // 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争
26.         // 当 this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败
27.         // 当 this.await返回true，进入循环尝试获取锁
28.         CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
29.         try {
30.             subscribeFuture.get(time, TimeUnit.MILLISECONDS);
31.         } catch (TimeoutException e) {
32.             if (!subscribeFuture.cancel(false)) {
33.                 subscribeFuture.whenComplete((res, ex) -> {
34.                     if (ex == null) {
35.                         unsubscribe(res, threadId);
36.                     }
37.                 });
38.             }
39.             acquireFailed(waitTime, unit, threadId);
40.             return false;
41.         } catch (ExecutionException e) {
42.             acquireFailed(waitTime, unit, threadId);
43.             return false;
44.         }
46.         try {
47.             time -= System.currentTimeMillis() - current
48.             // 超时，拿锁失败 返回false;
49.             if (time <= 0) {
50.                 acquireFailed(waitTime, unit, threadId);
51.                 return false;
52.             }
53.             // 自旋获取锁
54.             while (true) {
55.                 long currentTime = System.currentTimeMillis();
56.                 // 再次加锁
57.                 ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
58.                 // lock acquired
59.                 // 获得锁
60.                 if (ttl == null) {
61.                     return true;
62.                 }
64.                 time -= System.currentTimeMillis() - currentTime;
65.                 // 超时，拿锁失败 返回false;
66.                 if (time <= 0) {
67.                     acquireFailed(waitTime, unit, threadId);
68.                     return false;
69.                 }
71.                 // waiting for message
72.                 currentTime = System.currentTimeMillis();
73.                 // 下面的阻塞会在释放锁的时候，通过订阅发布及时relase
74.                 if (ttl >= 0 && ttl < time) {
75.                     // 如果锁的超时时间小于等待时间，通过SemaphorerelaseryAcquire阻塞锁的释放时间
76.                     commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
77.                 } else {
78.                     // 否则，通过Semaphore的tryAcquire阻塞传入的最大等待时间
79.                     commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
80.                 }
82.                 time -= System.currentTimeMillis() - currentTime;
83.                 if (time <= 0) {
84.                     acquireFailed(waitTime, unit, threadId);
85.                     return false;
86.                 }
87.             }
88.         } finally {
89.             // 取消订阅
90.             unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
91.         }
92.     }

复制代码

实行获取锁

1.     private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
2.         return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
3.     }
5.     private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
6.       RFuture<Long> ttlRemainingFuture;
7.       if (leaseTime > 0) {
8.         // 释放时间同步
9.         ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
10.       } else {
11.         // 如果没有传入锁的释放时间，默认 internalLockLeaseTime = 30000 MS
12.         ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
13.                 TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
14.       }
15.       CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
16.         // lock acquired
17.         // 如果返回null说明抢到了锁或者是可重入 否则直接返回还有多久过期
18.         if (ttlRemaining == null) {
19.           if (leaseTime > 0) {
20.             // 释放时间 赋值给 internalLockLeaseTime
21.             internalLockLeaseTime = unit.toMillis(leaseTime);
22.           } else {
23.             scheduleExpirationRenewal(threadId);
24.           }
25.         }
26.         // 没有抢到直接返回
27.         return ttlRemaining;
28.       });
29.       return new CompletableFutureWrapper<>(f);
30.     }
32.     <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
33.       return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
34.               
35.               "if (redis.call('exists', KEYS[1]) == 0) then " +
36.                       "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
37.                       "redis.call('pexpire', KEYS[1], ARGV[1]); " +
38.                       "return nil; " +
39.                       "end; " +
40.                       "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
41.                       "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
42.                       "redis.call('pexpire', KEYS[1], ARGV[1]); " +
43.                       "return nil; " +
44.                       "end; " +
45.                       "return redis.call('pttl', KEYS[1]);",
46.               Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
47.     }

复制代码

LUA脚天职析

1. # 加锁
2. "if (redis.call('exists', KEYS[1]) == 0) then " # 判断我的锁是否存在 =0 为不存在 没人抢占锁
3. "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + # 设置锁并计数重入次数
4. "redis.call('pexpire', KEYS[1], ARGV[1]); " + # 设置过期时间 30S
5. "return nil; " +
6. "end; " +
7. # 重入
8. "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + # 进入该逻辑说明有线程抢占了锁 继续判断是否同一个线程 ==1 为同一线程
9. "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + # 重入次数 + 1
10. "redis.call('pexpire', KEYS[1], ARGV[1]); " + # 设置超时时间
11. "return nil; " +
12. "end; " +
13. "return redis.call('pttl', KEYS[1]);", // 前面2个if都没进，说明锁被抢占并且不是同一线程，直接返回过期时间

复制代码

3、订阅

订阅锁状态，挂起叫醒线程
org.redisson.RedissonLock#subscribe

1. public CompletableFuture<E> subscribe(String entryName, String channelName) {
2.         AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
3.         CompletableFuture<E> newPromise = new CompletableFuture<>();
5.         semaphore.acquire().thenAccept(c -> {
6.             if (newPromise.isDone()) {
7.                 semaphore.release();
8.                 return;
9.             }
11.             E entry = entries.get(entryName);
12.             if (entry != null) {
13.                 entry.acquire();
14.                 semaphore.release();
15.                 entry.getPromise().whenComplete((r, e) -> {
16.                     if (e != null) {
17.                         newPromise.completeExceptionally(e);
18.                         return;
19.                     }
20.                     newPromise.complete(r);
21.                 });
22.                 return;
23.             }
25.             E value = createEntry(newPromise);
26.             value.acquire();
28.             E oldValue = entries.putIfAbsent(entryName, value);
29.             if (oldValue != null) {
30.                 oldValue.acquire();
31.                 semaphore.release();
32.                 oldValue.getPromise().whenComplete((r, e) -> {
33.                     if (e != null) {
34.                         newPromise.completeExceptionally(e);
35.                         return;
36.                     }
37.                     newPromise.complete(r);
38.                 });
39.                 return;
40.             }
42.             // 创建监听，释放锁，会发送消息
43.             RedisPubSubListener<Object> listener = createListener(channelName, value);
44.             CompletableFuture<PubSubConnectionEntry> s = service.subscribeNoTimeout(LongCodec.INSTANCE, channelName, semaphore, listener);
45.             newPromise.whenComplete((r, e) -> {
46.                 if (e != null) {
47.                     s.completeExceptionally(e);
48.                 }
49.             });
50.             s.whenComplete((r, e) -> {
51.                 if (e != null) {
52.                     value.getPromise().completeExceptionally(e);
53.                     return;
54.                 }
55.                 value.getPromise().complete(value);
56.             });
58.         });
60.         return newPromise;
61.     }

复制代码

org.redisson.pubsub.PublishSubscribe#createListener

1.     private RedisPubSubListener<Object> createListener(String channelName, E value) {
2.         // 创建监听，当监听到消息回来的时候，进入onMessage进行处理
3.         RedisPubSubListener<Object> listener = new BaseRedisPubSubListener() {
5.             @Override
6.             public void onMessage(CharSequence channel, Object message) {
7.                 if (!channelName.equals(channel.toString())) {
8.                     return;
9.                 }
11.                 PublishSubscribe.this.onMessage(value, (Long) message);
12.             }
13.         };
14.         return listener;
15.     }

复制代码

org.redisson.pubsub.LockPubSub#onMessage

1.     @Override
2.     protected void onMessage(RedissonLockEntry value, Long message) {
3.         if (message.equals(UNLOCK_MESSAGE)) {
4.             Runnable runnableToExecute = value.getListeners().poll();
5.             if (runnableToExecute != null) {
6.                 runnableToExecute.run();
7.             }
8.             // 释放 Semaphore
9.             value.getLatch().release();
10.         } else if (message.equals(READ_UNLOCK_MESSAGE)) {
11.             while (true) {
12.                 Runnable runnableToExecute = value.getListeners().poll();
13.                 if (runnableToExecute == null) {
14.                     break;
15.                 }
16.                 runnableToExecute.run();
17.             }
19.             value.getLatch().release(value.getLatch().getQueueLength());
20.         }
21.     }

复制代码

4、锁续期

redisson watchDog 使用时间轮技能，请参考时间轮算法分析
org.redisson.RedissonBaseLock#scheduleExpirationRenewal

1.     protected void scheduleExpirationRenewal(long threadId) {
2.         ExpirationEntry entry = new ExpirationEntry();
3.         // 放入EXPIRATION_RENEWAL_MAP 这个MAP中
4.         ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
5.         // 第一次进来，里面没有
6.         if (oldEntry != null) {
7.             // 如果其他线程来抢占这个锁，进入将线程ID保存至ExpirationEntry的threadIds这个Map中
8.             oldEntry.addThreadId(threadId);
9.         } else {
10.             // 将线程ID保存至ExpirationEntry的threadIds这个Map中
11.             entry.addThreadId(threadId);
12.             try {
13.                 // 执行
14.                 renewExpiration();
15.             } finally {
16.                 if (Thread.currentThread().isInterrupted()) {
17.                     cancelExpirationRenewal(threadId);
18.                 }
19.             }
20.         }
21.     }

复制代码

org.redisson.RedissonBaseLock#renewExpiration

1.   private void renewExpiration() {
2.         ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
3.         if (ee == null) {
4.             return;
5.         }
6.         // 开启时间轮，时间是10s之后执行我们的TimerTask任务
7.         Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
8.             @Override
9.             public void run(Timeout timeout) throws Exception {
10.                 // 从EXPIRATION_RENEWAL_MAP中拿到锁的对象，有可能在定时的时候被移除取消
11.                 ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
12.                 if (ent == null) {
13.                     return;
14.                 }
15.                 Long threadId = ent.getFirstThreadId();
16.                 if (threadId == null) {
17.                     return;
18.                 }
19.                 // 给锁续期
20.                 CompletionStage<Boolean> future = renewExpirationAsync(threadId);
21.                 future.whenComplete((res, e) -> {
22.                     // 异常报错，从Map移除
23.                     if (e != null) {
24.                         log.error("Can't update lock " + getRawName() + " expiration", e);
25.                         EXPIRATION_RENEWAL_MAP.remove(getEntryName());
26.                         return;
27.                     }
28.                     // 如果返回的是1 代表线程还占有锁，递归调用自己
29.                     if (res) {
30.                         // 递归再次加入时间轮
31.                         // reschedule itself
32.                         renewExpiration();
33.                     } else {
34.                         // 所不存在，这取消任务，移除相关MAP信息
35.                         cancelExpirationRenewal(null);
36.                     }
37.                 });
38.             }
39.         }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
40.         // 设置Timeout
41.         ee.setTimeout(task);
42.     }

复制代码

org.redisson.connection.MasterSlaveConnectionManager#newTimeout

1.     private HashedWheelTimer timer;
2.     @Override
3.     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
4.         try {
5.             // 添加进入时间轮
6.             return timer.newTimeout(task, delay, unit);
7.         } catch (IllegalStateException e) {
8.             if (isShuttingDown()) {
9.                 return DUMMY_TIMEOUT;
10.             }
11.             
12.             throw e;
13.         }
14.     }

复制代码

1.     protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
2.         return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
3.                 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + // 如果当前锁存在
4.                         "redis.call('pexpire', KEYS[1], ARGV[1]); " +  // 续期
5.                         "return 1; " +
6.                         "end; " +
7.                         "return 0;",
8.                 Collections.singletonList(getRawName()),
9.                 internalLockLeaseTime, getLockName(threadId));
10.     }

复制代码

5、开释锁

org.redisson.RedissonLock#unlock

1.     @Override
2.     public void unlock() {
3.         try {
4.             get(unlockAsync(Thread.currentThread().getId()));
5.         } catch (RedisException e) {
6.             if (e.getCause() instanceof IllegalMonitorStateException) {
7.                 throw (IllegalMonitorStateException) e.getCause();
8.             } else {
9.                 throw e;
10.             }
11.         }
12.     }

复制代码

org.redisson.RedissonBaseLock.unlockAsync

1.   @Override
2.   public RFuture<Void> unlockAsync(long threadId) {
3.     // 进入释放锁逻辑
4.     RFuture<Boolean> future = unlockInnerAsync(threadId);
6.     CompletionStage<Void> f = future.handle((opStatus, e) -> {
7.       // 移除ExpirationEntry中的threadId 并且移除 EXPIRATION_RENEWAL_MAP中的ExpirationEntry watchDog就不会继续续期  
8.       cancelExpirationRenewal(threadId);
9.       
10.       // 异常处理
11.       if (e != null) {
12.         throw new CompletionException(e);
13.       }
14.       // 不存在锁信息
15.       if (opStatus == null) {
16.         IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
17.                 + id + " thread-id: " + threadId);
18.         throw new CompletionException(cause);
19.       }
21.       return null;
22.     });
24.     return new CompletableFutureWrapper<>(f);
25.   }

复制代码

org.redisson.RedissonBaseLock.unlockInnerAsync

1. protected RFuture<Boolean> unlockInnerAsync(long threadId) {
2.     return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
3.             "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
4.             "return nil;" +
5.             "end; " +
6.             "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
7.             "if (counter > 0) then " +
8.             "redis.call('pexpire', KEYS[1], ARGV[2]); " +
9.             "return 0; " +
10.             "else " +
11.             "redis.call('del', KEYS[1]); " +
12.             "redis.call('publish', KEYS[2], ARGV[1]); " +
13.             "return 1; " +
14.             "end; " +
15.             "return nil;",
16.     Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
17. }

复制代码

开释锁LUA

1. "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +   # 锁已经释放
2.             "return nil;" +
3.             "end; " +
4.             "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +  #重入次数减一
5.             "if (counter > 0) then " +
6.             "redis.call('pexpire', KEYS[1], ARGV[2]); " + # 重入次数还不为0 那说明占有锁，设置过期时间
7.             "return 0; " +
8.             "else " +
9.             "redis.call('del', KEYS[1]); " +    # 重入次数为0，释放锁
10.             "redis.call('publish', KEYS[2], ARGV[1]); " + # 发布订阅事件，唤醒其它线程，可以去竞争锁了
11.             "return 1; " +
12.             "end; " +
13.             "return nil;",

复制代码

6、流程图

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

继续阅读请点击广告