ToB企服应用市场:ToB评测及商务社交产业平台

标题: 如何实现一个分布式锁 [打印本页]

---

作者: 金歌    时间: 2024-7-13 20:04
标题: 如何实现一个分布式锁
如何实现一个分布式锁

本篇内容主要先容如何使用 Java 语言实现一个注解式的分布式锁，主要是通过注解+AOP 环绕通知来实现。
1. 锁注解

我们首先写一个锁的注解

1. /**
2. * 分布式锁注解
3. */
4. @Retention(RetentionPolicy.RUNTIME)
5. @Target({ElementType.METHOD})
6. @Documented
7. public @interface RedisLock {
9.         long DEFAULT_TIMEOUT_FOR_LOCK = 5L;
10.         long DEFAULT_EXPIRE_TIME = 60L;
12.         String key() default "your-biz-key";
14.         long expiredTime() default DEFAULT_EXPIRE_TIME;
16.         long timeoutForLock() default DEFAULT_TIMEOUT_FOR_LOCK;
18. }

复制代码

expiredTime 是设置锁的逾期时间，timeoutForLock 是设置等待锁的超时时间。如果没有等待得到锁的超时时间这个功能，那么其他线程在获取锁失败时只能直接失败，无法进行排队等待。
我们如何使用这个注解呢，很轻易，在需要加锁的业务方法上直接用就行.如下，我们有一个库存服务类，它有一个扣减库存方法，该方法将数据库中的一个库存商品的数量减一。在并发场景下，如果我们没有对其进行资源控制，必然会发生库存扣减差别等征象。

1. public class StockServiceImpl {
2.         @RedisLock(key = "stock-lock", expiredTime = 10L, timeoutForLock = 5L)
3.         public void deduct(Long stockId) {
4.                 Stock stock = this.getById(1L);
5.                 Integer count = stock.getCount();
6.                 stock.setCount(count - 1);
7.                 this.updateById(stock);
8.         }
9. }

复制代码

2. 在 AOP 切面中进行加锁处置处罚

我们需要使用 AOP 来处置处罚什么？自然是处置处罚使用@RedisLock的方法，因此我们写一个切点表达式，它匹配所有标有 @RedisLock 注解的方法。
接着，我们将此切点表达式与 @Around 注解结合使用，以创建环绕通知，在目标方法执行前后执行我们的加锁解锁逻辑。
因此，根本的逻辑我们就理清了，代码大致长下面这个样子:

1. public class RedisLockAspect {
3.         private final RedisTemplate<String, Object> redisTemplate;
5.         // 锁的redis key前缀
6.         private static final String DEFAULT_KEY_PREFIX = "lock:";
8.         // 匹配所有标有 @RedisLock 注解的方法
9.         @Pointcut("@annotation(com.kelton.lock.annotation.RedisLock)")
10.         public void lockAnno() {
11.         }
14.         @Around("lockAnno()")
15.         public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
16.                 // 获取拦截方法上的RedisLock注解
17.                 RedisLock annotation = getLockAnnotationOnMethod(joinPoint);
18.                 // 获取锁key
19.                 String key = getKey(annotation);
20.                 // 锁过期时间
21.                 long expireTime = annotation.expiredTime();
22.                 // 获取锁的等待时间
23.                 long timeoutForLock = annotation.timeoutForLock();
24.                 // 在这里加锁
25.                 someCodeForLock...
26.                 // 执行业务
27.                 joinPoint.proceed();
28.                 // 在这里解锁
29.                 someCodeForUnLock...
30.         }

复制代码

我们在加锁的时候，需要用上 timeoutForLock 这个属性，我们通过自旋加线程休眠的方式，来达到在一段时间内等待获取锁的目的。如果自旋时间结束后，还没获取锁，则抛出异常，这里可以根据自己情况而定。自旋加锁代码如下：

1.     // 自旋获取锁
2.     long endTime = System.currentTimeMillis() + timeoutForLock * 1000;
3.     boolean acquired = false;
4.     String uuid = UUID.randomUUID().toString();
5.     while(System.currentTimeMillis() < endTime) {
6.         Boolean absent = redisTemplate.opsForValue()
7.                 .setIfAbsent(key, uuid, expireTime, TimeUnit.SECONDS);
9.         if (Boolean.TRUE.equals(absent)) {
10.             acquired = true;
11.             break;
12.         } else {
13.             // 获取不到锁，尝试休眠100毫秒后重试
14.             Thread.sleep(100);
15.         }
16.     }
17.     // 超时未获取到锁， 抛出异常，可根据自己业务而定
18.     if (!acquired) {
19.         throw new RuntimeException("获取锁异常");
20.     }

复制代码

我们发现上面加锁的时候设置了一个 uuid 作为 value 值，这是为了在锁释放的时候，不误删其他线程上的锁，随后，我们就可以执行被 AOP 切中的方法，执行结束释放锁。代码如下：

1.     try {
2.         // 执行业务
3.         joinPoint.proceed();
4.     } catch (Throwable e) {
5.         log.error("业务执行出错！");
6.     } finally {
7.         // 解锁时进行校验，只删除自己线程加的锁
8.         String value = (String) redisTemplate.opsForValue().get(key);
9.         if (uuid.equals(value)) {
10.             redisTemplate.delete(key);
11.         } else {
12.             log.warn("锁已过期！");
13.         }
14.     }

复制代码

到这里，我们就以注解+AOP 的方式实现了分布式锁的功能。当然，以上只实现了分布式锁的简单功能，还缺少了分布式锁的 key 自动续约防止锁逾期功能，以及锁重入功能。
目前，RedisLockAspect的完整代码如下：

1. @Component
2. @Aspect
3. @Slf4j
4. @AllArgsConstructor
5. public class RedisLockAspect {
7.         // 匹配所有标有 @RedisLock 注解的方法
8.         @Pointcut("@annotation(com.kelton.lock.annotation.RedisLock)")
9.         public void lockAnno() {
10.         }
13.         @Around("lockAnno()")
14.         public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
15.                 // 获取拦截方法上的RedisLock注解
16.                 RedisLock annotation = getLockAnnotationOnMethod(joinPoint);
18.                 String key = getKey(annotation);
19.                 // 锁过期时间
20.                 long expireTime = annotation.expiredTime();
21.                 // 获取锁的等待时间
22.                 long timeoutForLock = annotation.timeoutForLock();
23.                 // 自旋获取锁
24.                 long endTime = System.currentTimeMillis() + timeoutForLock * 1000;
25.                 boolean acquired = false;
26.                 String uuid = UUID.randomUUID().toString();
27.                 while(System.currentTimeMillis() < endTime) {
28.                         Boolean absent = redisTemplate.opsForValue()
29.                                         .setIfAbsent(key, uuid, expireTime, TimeUnit.SECONDS);
31.                         if (Boolean.TRUE.equals(absent)) {
32.                                 acquired = true;
33.                                 break;
34.                         } else {
35.                                 // 获取不到锁，尝试休眠100毫秒后重试
36.                                 Thread.sleep(100);
37.                         }
38.                 }
39.                 // 超时未获取到锁， 抛出异常，可根据自己业务而定
40.                 if (!acquired) {
41.                         throw new RuntimeException("获取锁异常");
42.                 }
43.                 try {
44.                         // 执行业务
45.                         joinPoint.proceed();
46.                 } catch (Throwable e) {
47.                         log.error("业务执行出错！");
48.                 } finally {
49.                         // 解锁时进行校验，只删除自己线程加的锁
50.                         String value = (String) redisTemplate.opsForValue().get(key);
51.                         if (uuid.equals(value)) {
52.                                 redisTemplate.delete(key);
53.                         } else {
54.                                 log.warn("锁已过期！");
55.                         }
56.                 }
57.         }
59.         private String getKey(RedisLock redisLock) {
60.                 if (Objects.isNull(redisLock)) {
61.                         return DEFAULT_KEY_PREFIX + "default";
62.                 }
63.                 return DEFAULT_KEY_PREFIX + redisLock.key();
64.         }
66.         private RedisLock getLockAnnotationOnMethod(ProceedingJoinPoint joinPoint) {
67.                 MethodSignature signature = (MethodSignature) joinPoint.getSignature();
68.                 Method method = signature.getMethod();
69.                 return method.getAnnotation(RedisLock.class);
70.         }
72. }

复制代码

3. key 自动续约防止锁逾期

我们接着完善该分布式锁，为其添加 key 自动续约防止锁逾期的功能。我们的思路与Redission的watch dog类似，开启一个后台线程，来定时查抄需要续约的锁。我们如何判断一个锁是否需要续约呢，我们可以简单定义一个续约分界线，比如在锁逾期时间的三分之二的时间点及之后，对锁进行续约。
3.1 定义一个续约任务4

我们来定义一个锁续约任务，那我们需要什么信息呢？
我们至少需要锁的 key，锁要设置的逾期时间。这是两个最根本的信息。
要判断在锁逾期时间的三分之二的时间点及之后进行续约，那么我们还需要纪录锁上次续约的时间点。
此外，我们还可以为锁续约任务添加最大续约次数限制，这可以避免某些执行时间特别久的任务不断占用锁。所以我们还需要纪录当前锁续约次数和最大续约次数。
对超过最大续约次数的锁的线程，我们直接将其制止，因此我们也纪录一下该锁的线程。
结合上面的分析，我们定义的锁续约任务类如下：

1. public class LockRenewTask {
3.         /**
4.          * key
5.          */
6.         private final String key;
7.         /**
8.          * 过期时间。单位：秒
9.          */
10.         private final long expiredTime;
11.         /**
12.          * 锁的最大续约次数
13.          */
14.         private final int maxRenewCount;
15.         /**
16.          * 锁的当前续约次数
17.          */
18.         private int currentRenewCount;
19.         /**
20.          * 最新更新时间
21.          */
22.         private LocalDateTime latestRenewTime;
23.         /**
24.          * 业务线程
25.          */
26.         private final Thread thread;
28.         public LockRenewTask(String key, long expiredTime, int maxRenewCount, Thread thread) {
29.                 this.key = key;
30.                 this.expiredTime = expiredTime;
31.                 this.maxRenewCount = maxRenewCount;
32.                 this.thread = thread;
33.                 this.latestRenewTime = LocalDateTime.now();
34.         }
35.         /**
36.          * 是否到达续约时间
37.          * @return
38.          */
39.         public boolean isTimeToRenew() {
40.                 LocalDateTime now = LocalDateTime.now();
41.                 Duration duration = Duration.between(latestRenewTime, now);
43.                 return duration.toSeconds() >= ((double)(this.expiredTime / 3) * 2);
44.         }
45.         /**
46.          * 是否达到最大续约次数
47.          * @return
48.          */
49.         public boolean exceedMaxRenewCount() {
50.                 return this.currentRenewCount >= this.maxRenewCount;
51.         }
52.         public synchronized void renew() {
53.                 this.currentRenewCount++;
54.                 this.latestRenewTime = LocalDateTime.now();
55.         }
56.         // 取消业务方法
57.         public void cancel() {
58.                 thread.interrupt();
59.         }
60.         public String getKey() {
61.                 return key;
62.         }
63.         public long getExpiredTime() {
64.                 return expiredTime;
65.         }
66. }

复制代码

我们添��了一些关于锁续约的方法：

• isTimeToRenew(): 判断是否可以对锁进行续约
  
• exceedMaxRenewCount(): 判断是否达到最大续约次数
  
• renew(): 来标记一次续约操纵
  
• cancel(): 取消业务方法
  

3.2 定义一个锁续约任务处置处罚器

接着，我们定义一个定时执行该续约任务的 handler。该 handler 也比较简答，核心逻辑是持有一个类型为 List的 taskList 来添加续约任务，且使用一个 ScheduledExecutorService 来定时遍历该 taskList 来执行续约任务。该 handler 再对外暴露一个 addRenewTask 方法，方便外部调用来添加续约任务到 taskList 中。

1. @Slf4j
2. @Component
3. public class LockRenewHandler {
5.         @Autowired
6.         private RedisTemplate<String, Object> redisTemplate;
8.         /**
9.          * 保障对 taskList的添加删除操作是线程安全的
10.          */
11.         private final ReentrantLock taskListLock = new ReentrantLock();
13.         private final List<LockRenewTask> taskList = new ArrayList<>();
15.         private final ScheduledExecutorService taskExecutorService;
17.         {
18.                 taskExecutorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
19.                 taskExecutorService.scheduleAtFixedRate(() -> {
20.                         try {
21.                                 executeRenewTask();
22.                         } catch (Exception e) {
23.                                 //错误处理
24.                         }
25.                 }, 1, 2, TimeUnit.SECONDS);
27.         }
28.         /**
29.          * 添加续约任务
30.          */
31.         public void addRenewTask(LockRenewTask task) {
32.                 taskListLock.lock();
33.                 try {
34.                         taskList.add(task);
35.                 } finally {
36.                         taskListLock.unlock();
37.                 }
38.         }
39.         /**
40.          * 执行续约任务
41.          */
42.         private void executeRenewTask() {
43.                 log.info("开始执行续约任务");
44.                 if (CollectionUtils.isEmpty(taskList)) {
45.                         return;
46.                 }
47.                 // 需要删除的任务，暂存这个集合中  取消
48.                 List<LockRenewTask> cancelTask = new ArrayList<>();
49.                 // 获取任务副本
50.                 List<LockRenewTask> copyTaskList = new ArrayList<>(taskList);
51.                 for (LockRenewTask task : copyTaskList) {
52.                         try {
53.                                 // 判断 Redis 中是否存在 key
54.                                 if (!redisTemplate.hasKey(task.getKey())) {
55.                                         cancelTask.add(task);
56.                                         continue;
57.                                 }
58.                                 // 大于等于最大续约次数
59.                                 if (task.exceedMaxRenewCount()) {
60.                                         // 停止续约任务
61.                                         task.cancel();
62.                                         cancelTask.add(task);
63.                                         continue;
64.                                 }
65.                                 // 到达续约时间
66.                                 if (task.isTimeToRenew()) {
67.                                         log.info("续约任务：{}", task.getKey());
68.                                         redisTemplate.expire(task.getKey(), task.getExpiredTime(), TimeUnit.SECONDS);
69.                                         task.renew();
70.                                 }
71.                         } catch (Exception e) {
72.                                 //错误处理
73.                                 log.error("处理任务出错：{}", task);
74.                         }
75.                 }
76.                 // 加锁，删除 taskList 中需要移除的任务
77.                 taskListLock.lock();
78.                 try {
79.                         taskList.removeAll(cancelTask);
80.                         // 清理cancelTask,避免堆积，产生内存泄露
81.                         cancelTask.clear();
82.                 } finally {
83.                         taskListLock.unlock();
84.                 }
85.         }
86. }

复制代码

总结一下 LockRenewHandler的主要作用：它负责管理和执行续约任务，以延长 Redis 中键的逾期时间。

• 添加续约任务：addRenewTask() 方法允许添加新的续约任务到内部列表 taskList 中。
  
• 执行续约任务：executeRenewTask() 方法定期执行续约任务。它查抄每个任务的状态，并根据需要续约 Redis 中的键。
  
• 移除完成的任务：维护一个 cancelTask 列表，用于存储需要从 taskList 中移除的任务。在 executeRenewTask() 方法中，它会将完成的任务添加到 cancelTask 列表中，并在之后将其从 taskList 中移除。
  

大概的工作流程如下：

• 续约任务被添加到 taskList 中。
  
• executeRenewTask() 方法定期执行，它查抄每个任务的状态：
  
  
  
  • 如果 Redis 中不再存在该键，则取消任务。
    
  • 如果任务的续约次数达到上限，则取消任务。
    
  • 如果是时候续约了，则续约 Redis 中的键并更新任务的续约次数，纪录续约时间点。
    
  
  
• 完成的任务被添加到 cancelTask 列表中。
  
• executeRenewTask() 方法获取 taskList 的副本，并从副本中移除 cancelTask 中的任务，并且在完成移除任务操纵后清空cancelTask。
  
• 更新后的 taskList 被生存回类中。
  

两个需要注意的点

• 我们遍历taskList时拷贝了一份副本进行遍历，因为taskList是可变的，这样可以避免在遍历的时候产生并发修改问题。
  
• cancelTask需要清算，避免产生内存泄漏。
  

通过这种方式，LockRenewHandler 可以确保 Redis 中的键在需要时得到续约，并自动移除完成或失败的任务。
3.3 添加锁续约任务

在上面 3.1 节和 3.2 节我们定义好了锁续约任务和处置处罚锁续约任务的核心代码，接下来我们需要在第 2 节加锁解锁的 AOP 处置处罚逻辑上进行一点小小的修改，主要就是在执行加锁之后，执行业务代码之前，添加上锁续约任务。修改位置如下：

1. public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
2.     ... // 省略代码
3.     try {
4.         // 添加锁续约任务
5.         LockRenewTask task = new LockRenewTask(key, annotation.expiredTime(), annotation.maxRenew(), Thread.currentThread());
6.         lockRenewHandler.addRenewTask(task);
7.         log.info("添加续约任务, key:{}", key);
8.         // 执行业务
9.         joinPoint.proceed();
10.     } catch (Throwable e) {
11.         log.error("业务执行出错！");
12.     } finally {
13.         // 解锁时进行校验，只删除自己线程加的锁
14.         String value = (String) redisTemplate.opsForValue().get(key);
15.         if (uuid.equals(value)) {
16.             redisTemplate.delete(key);
17.         } else {
18.             log.warn("锁已过期！");
19.         }
20.     }
21.     ... // 省略代码
22. }

复制代码

到这里，我们的分布式锁已经相当完善了，把锁自动续约的功能也加上了。当然，还没有实现锁的可重入性。

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

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4