拒绝躺平，如何使用AOP的环绕通知实现分布式锁

个人博客

https://zhaoxiaobin.net

如何在分布式环境下，像用synchronized关键字那样使用分布式锁。比如开发一个注解，叫@DistributionLock，作用于一个方法函数上，每次调方法前加锁，调完之后自动释放锁。

可以利用Spring AOP中环绕通知的特性，完全满足上面的要求。

1、Maven依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.11.5</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
</dependencies>

这里采用redisson这个第三方库来做redis的分布式锁，redis配置不在这里展开，可以参考文末贴出的代码地址或具体可以看这篇 https://zhaoxiaobin.net/6ba1dd4958f4/

2、开发自定义注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributionLock {
    /**
     * 分布式锁key
     */
    String value() default "";

    /**
     * 获取分布式锁的等待时间
     */
    int waitTime() default 5 * 1000;

    /**
     * 分布式锁key所在参数列表中的位置
     */
    int index() default -1;
}

这里按照锁的粗细粒度分为两种模式：

粗粒度：由注解中的value字段指定，编译阶段就确定了，同一个方法（业务）共享该锁。不管是谁调这个方法，都是按串行执行。
细粒度：方法的参数列表中的一个参数作为锁的key值，比如一个编号、一个流水号等等业务唯一参数。主要应用在如果同一个（同一组）交易允许不同人同时做，但同一个人必须串行执行的场景；由index指定作为key的形参位置。

3、开发切面

@Order(Integer.MIN_VALUE + 1)
@Aspect
@Component
@Slf4j
public class DistributionLockAspect {
    @Autowired
    private RedissonClient redissonClient;

    @Around("@annotation(distributionLock)")
    public Object doAround(ProceedingJoinPoint point, DistributionLock distributionLock) throws Throwable {
        String methodName = point.getSignature().getName();
        if (StringUtils.isNotBlank(distributionLock.value())) {
            // 锁粒度较粗，由目标方法上的DistributionLock注解中指定锁的名称，由同一个方法（业务）共享该锁
            return this.tryLock(point, distributionLock.value(), distributionLock.waitTime());
        } else if (distributionLock.index() >= 0){
            // 锁粒度较细，由目标方法的第x个参数作为锁名称，可以是一个业务上的编号、名称等等
            Object[] args = point.getArgs(); // 参数列表
            int index = distributionLock.index(); // 参数列表中第几个参数作为分布式锁的key
            if (args.length <= index) {
                log.error("目标方法:{}上没有第:{}参数", methodName, index);
                throw new RuntimeException("目标方法:" + methodName + "上没有第:" + index + "参数");
            }
            String key = args[index].toString();
            return this.tryLock(point, key, distributionLock.waitTime());
        } else {
            log.error("没有配置具体分布式锁的key,目标方法:{}", methodName);
            throw new RuntimeException("没有配置具体分布式锁的key,目标方法:" + methodName);
        }
    }

    /**
     * 只获取、释放锁，不处理任何异常，原样抛出异常
     *
     * @param point    切入点
     * @param key      分布式锁的key
     * @param waitTime 获取锁等待时间
     * @return
     * @throws Throwable
     */
    private Object tryLock(ProceedingJoinPoint point, String key, int waitTime) throws Throwable {
        RLock disLock = redissonClient.getLock(key);
        // 默认30秒后自动过期，每隔30/3=10秒，看门狗（守护线程）会去续期锁，重设为30秒
        boolean tryLock = disLock.tryLock(waitTime, TimeUnit.MILLISECONDS);
        if (!tryLock) {
            // 由具体业务决定是抛异常还是返回null或其他业务对象
//            throw new RuntimeException("获取分布式锁失败");
            return false;
        } else {
            try {
                return point.proceed(point.getArgs());
            } finally {
                disLock.unlock();
            }
        }
    }
}

4、模拟测试扣库存

业务操作

@Service
public class DistributionLockDemo {
    /**
     * 库存
     */
    public static int count = 20;

    public boolean increment() {
        if (count > 0) {
            count--;
            return true;
        }
        return false;
    }

    public int get() {
        return count;
    }
}

模拟发起交易

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class DistributionLockTests {
    @Autowired
    private DistributionLockDemo distributionLockDemo;

    /**
     * 多线程并行处理500个请求，扣库存
     */
    @Test
    public void testDistributionLock() {
        long successCount = IntStream.range(0, 500).parallel().filter(j -> distributionLockDemo.increment()).count();

        log.info("成功数:{}", successCount);
        log.info("剩余库存:{}", distributionLockDemo.get());
    }
}

这里用Java8的并行流并发去发50个交易来模拟扣20的库存，不加锁的情况下：

1
2

2021-06-30 18:10:20,942 [INFO] [main] [net.zhaoxiaobin.redisson.DistributionLockTests:35] [] 成功数:27
2021-06-30 18:10:20,942 [INFO] [main] [net.zhaoxiaobin.redisson.DistributionLockTests:36] [] 剩余库存:0

给increment方法加上分布式锁注解，再测试：

@DistributionLock(value = "incrementLock", waitTime = 1000)
public boolean increment() {
    if (count > 0) {
        count--;
        return true;
    }
    return false;
}

1
2

2021-06-30 18:11:30,722 [INFO] [main] [net.zhaoxiaobin.redisson.DistributionLockTests:35] [] 成功数:20
2021-06-30 18:11:30,722 [INFO] [main] [net.zhaoxiaobin.redisson.DistributionLockTests:36] [] 剩余库存:0

以上案例不是太严谨，为了测试方便，只是测了单机并没有测集群的效果。有条件可以连上数据库并起多个服务去扣库里的数据，看分布式锁效果如何。

5、分布式锁的安全性

相信有不少同学都知道Redis的分布式锁不是那么的万无一失；比如主从切换导致锁丢失，还有NPC等问题影响锁的安全性，具体可以参考这篇 https://zhaoxiaobin.net/89e72f2a4185/

建议：

对于要求数据绝对正确的业务，在资源层一定要做好「兜底」，比如数据库的乐观锁、类似CAS等操作。
使用分布式锁，在上层完成「互斥」目的，虽然极端情况下锁会失效，但它可以最大程度把并发请求阻挡在最上层，减轻操作资源层的压力。