分布式锁-yudao之设计精妙之处

芋道源码ruoyi-vue-pro

源码地址：gitee

官方文档

B站视频教程

分布式锁的实现开始

/**
 * 支付通知的锁 Redis DAO
 *
 * @author 芋道源码
 */
@Repository
public class PayNotifyLockRedisDAO {

    @Resource
    private RedissonClient redissonClient;

    public void lock(Long id, Long timeoutMillis, Runnable runnable) {
        // 获取到当前的redis中的唯一key（pay_notify:lock:支付id）
        String lockKey = formatKey(id);
        // 获取锁
        RLock lock = redissonClient.getLock(lockKey);
        try {
            // 加锁并获取超时时间
            lock.lock(timeoutMillis, TimeUnit.MILLISECONDS);
            // 执行逻辑
            runnable.run();
        } finally {
            // 释放锁
            lock.unlock();
        }
    }
    private static String formatKey(Long id) {
        // 包装成 ：pay_notify:lock:支付id 返回
        return String.format(PAY_NOTIFY_LOCK.getKeyTemplate(), id);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

精妙：加锁payNotifyLockCoreRedisDAO.lock（）具体的业务逻辑通过：runnable.run(); 他来执行

业务层的方法调用

 /**
     * 同步执行单个支付通知
     *
     * @param task 通知任务
     */
    public void executeNotifySync(PayNotifyTaskDO task) {
        // 分布式锁，避免并发问题
        payNotifyLockCoreRedisDAO.lock(task.getId(), NOTIFY_TIMEOUT_MILLIS, () -> {
            // 校验，当前任务是否已经被通知过
            // 虽然已经通过分布式加锁，但是可能同时满足通知的条件，然后都去获得锁。此时，第一个执行完后，第二个还是能拿到锁，然后会再执行一次。
            PayNotifyTaskDO dbTask = payNotifyTaskCoreMapper.selectById(task.getId());
            if (DateUtils.afterNow(dbTask.getNextNotifyTime())) {
                log.info("[executeNotify][dbTask({}) 任务被忽略，原因是未到达下次通知时间，可能是因为并发执行了]", JsonUtils.toJsonString(dbTask));
                return;
            }

            // 执行通知
            executeNotify(dbTask);
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

给一个runnable 的执行 ：也就是业务逻辑。

在给大家介绍一个比较常用的JUC类

 @Override
    public int executeNotify() throws InterruptedException {
        // 获得需要通知的任务
        List<PayNotifyTaskDO> tasks = payNotifyTaskCoreMapper.selectListByNotify();
        if (CollUtil.isEmpty(tasks)) {
            return 0;
        }

        // 遍历，逐个通知
        CountDownLatch latch = new CountDownLatch(tasks.size());
        tasks.forEach(task -> threadPoolTaskExecutor.execute(() -> {
            try {
                executeNotifySync(task);
            } finally {
                latch.countDown();
            }
        }));
        // 等待完成
        awaitExecuteNotify(latch);
        // 返回执行完成的任务数（成功 + 失败)
        return tasks.size();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

CountDownLatch latch.countDown();减少锁存器的计数，如果计数达到零，则释放所有等待的线程。
如果当前计数大于零，则递减。如果新计数为零，则出于线程调度目的，将重新启用所有等待线程。
如果当前计数等于零，则不会发生任何事情。
awaitExecuteNotify(latch):等待全部支付通知的完成 每 1 秒会打印一次剩余任务数量
latch.await(1L, TimeUnit.SECONDS) 一秒后唤醒新的线程 一般不设置