<p><a href="https://blog.csdn.net/qq_33722836/article/details/117462068"><img alt="" src="/files/default/2021/06-24/09093715aa46406675.png"></a></p>

如何在分布式环境下,像用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配置不在这里展开,可以参考文末贴出的代码地址或具体可以看这篇 zhaoxiaobin.net/Redis/Redis…

2、开发自定义注解

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

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

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

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

  1. 粗粒度:由注解中的value字段指定,编译阶段就确定了,同一个方法(业务)共享该锁。不管是谁调这个方法,都是按串行执行。
  2. 细粒度:方法的参数列表中的一个参数作为锁的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);
        try {
            // 默认30秒后自动过期,每隔30/3=10秒,看门狗(守护线程)会去续期锁,重设为30秒
            boolean tryLock = disLock.tryLock(waitTime, TimeUnit.MILLISECONDS);
            if (!tryLock) {
                log.error("获取分布式锁:{}失败", key);
                // 由具体业务决定是抛异常还是返回null或其他业务对象
//                throw new RuntimeException("获取分布式锁失败");
                return false;
            }
            return point.proceed(point.getArgs());
        } finally {
            // 只有获取到锁的线程才执行释放锁操作
            if (disLock.isHeldByCurrentThread()) {
                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的库存,不加锁的情况下:

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;
}
复制代码
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等问题影响锁的安全性,具体可以参考这篇 zhaoxiaobin.net/Redis/Redis…

建议

  1. 但对于要求数据绝对正确的业务,在资源层一定要做好「兜底」,比如数据库的乐观锁、类似CAS等操作。
  2. 使用分布式锁,在上层完成「互斥」目的,虽然极端情况下锁会失效,但