如何在分布式环境下，像用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的分布式锁
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); 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等问题影响锁的安全性
建议：
- 但对于要求数据绝对正确的业务，在资源层一定要做好「兜底」，比如数据库的乐观锁、类似CAS等操作。
- 使用分布式锁，在上层完成「互斥」目的，虽然极端情况下锁会失效，但它可以最大程度把并发请求阻挡在最上层，减轻操作资源层的压力。
