基于注解优雅的实现接口幂等性

系统管理员 2024-03-24 00:08 70阅读 0赞

## **一、什么是幂等性** ##

简单来说，就是对一个接口执行重复的多次请求，与一次请求所产生的结果是相同的，听起来非常容易理解，但要真正的在系统中要始终保持这个目标，是需要很严谨的设计的，在实际的生产环境下，我们应该保证任何接口都是幂等的，而如何正确的实现幂等，就是本文要讨论的内容。

## **二、哪些请求天生就是幂等的** ##

首先，我们要知道查询类的请求一般都是天然幂等的，除此之外，删除请求在大多数情况下也是幂等的，但是ABA场景下除外。

**举一个简单的例子**

比如，先请求了一次删除A的操作，但由于响应超时，又自动请求了一次删除A的操作，如果在两次请求之间，又插入了一次A，而实际上新插入的这一次A，是不应该被删除的，这就是ABA问题，不过，在大多数业务场景中，ABA问题都是可以忽略的。

除了查询和删除之外，还有更新操作，同样的更新操作在大多数场景下也是天然幂等的，其例外是也会存在ABA的问题，更重要的是，比如执行`update table set a = a + 1 where v = 1`这样的更新就非幂等了。

最后，就还剩插入了，插入大多数情况下都是非幂等的，除非是利用数据库唯一索引来保证数据不会重复产生。

## **三、为什么需要幂等** ##

### **1.超时重试** ###

当发起一次RPC请求时，难免会因为网络不稳定而导致请求失败，一般遇到这样的问题我们希望能够重新请求一次，正常情况下没有问题，但有时请求实际上已经发出去了，只是在请求响应时网络异常或者超时，此时，请求方如果再重新发起一次请求，那被请求方就需要保证幂等了。

### **2.异步回调** ###

异步回调是提升系统接口吞吐量的一种常用方式，很明显，此类接口一定是需要保证幂等性的。

### **3.消息队列** ###

现在常用的消息队列框架，比如：Kafka、RocketMQ、RabbitMQ在消息传递时都会采取At least once原则（也就是至少一次原则，在消息传递时，不允许丢消息，但是允许有重复的消息），既然消息队列不保证不会出现重复的消息，那消费者自然要保证处理逻辑的幂等性了。

## **四、实现幂等的关键因素** ##

### **关键因素1** ###

幂等唯一标识，可以叫它幂等号或者幂等令牌或者全局ID，总之就是客户端与服务端一次请求时的唯一标识，一般情况下由客户端来生成，也可以让第三方来统一分配。

### **关键因素2** ###

有了唯一标识以后，服务端只需要确保这个唯一标识只被使用一次即可，一种常见的方式就是利用数据库的唯一索引。

## **五、注解实现幂等性** ##

1.  ###  **定义DistributedLock注解** ###
    
        @Target({ElementType.METHOD})
        @Retention(RetentionPolicy.RUNTIME)
        @Documented
        public @interface DistributedLock {
         
            /**
             * 保证业务接口的key的唯一性，否则失去了分布式锁的意义 锁key
             * 支持使用spEl表达式
             */
            String key();
         
            /**
             * 保证业务接口的key的唯一性，否则失去了分布式锁的意义 锁key 前缀
             */
            String keyPrefix() default "";
         
            /**
             * 是否在等待时间内获取锁，如果在等待时间内无法获取到锁，则返回失败
             */
            boolean tryLok() default false;
         
            /**
             * 获取锁的最大尝试时间 ，会尝试tryTime时间获取锁，在该时间内获取成功则返回，否则抛出获取锁超时异常，tryLok=true时，该值必须大于0。
             *
             */
            long tryTime() default 0;
         
            /**
             * 加锁的时间，超过这个时间后锁便自动解锁
             */
            long lockTime() default 30;
         
            /**
             * tryTime 和 lockTime的时间单位
             */
            TimeUnit unit() default TimeUnit.SECONDS;
         
            /**
             * 是否公平锁，false:非公平锁，true:公平锁
             */
            boolean fair() default false;
        }
2.  ### **定义DistributedLockAspect Lock切面** ###
    
        @Aspect
        @Slf4j
        public class DistributedLockAspect {
         
            @Resource
            private IDistributedLock distributedLock;
         
            /**
             * SpEL表达式解析
             */
            private SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
         
            /**
             * 用于获取方法参数名字
             */
            private DefaultParameterNameDiscoverer nameDiscoverer = new DefaultParameterNameDiscoverer();
         
            @Pointcut("@annotation(com.yt.bi.common.redis.distributedlok.annotation.DistributedLock)")
            public void distributorLock() {
            }
         
            @Around("distributorLock()")
            public Object around(ProceedingJoinPoint pjp) throws Throwable {
                // 获取DistributedLock
                DistributedLock distributedLock = this.getDistributedLock(pjp);
                // 获取 lockKey
                String lockKey = this.getLockKey(pjp, distributedLock);
                ILock lockObj = null;
                try {
                    // 加锁，tryLok = true,并且tryTime > 0时，尝试获取锁，获取不到超时异常
                    if (distributedLock.tryLok()) {
                        if(distributedLock.tryTime() <= 0){
                            throw new IdempotencyException("tryTime must be greater than 0");
                        }
                        lockObj = this.distributedLock.tryLock(lockKey, distributedLock.tryTime(), distributedLock.lockTime(), distributedLock.unit(), distributedLock.fair());
                    } else {
                        lockObj = this.distributedLock.lock(lockKey, distributedLock.lockTime(), distributedLock.unit(), distributedLock.fair());
                    }
         
                    if (Objects.isNull(lockObj)) {
                        throw new IdempotencyException("Duplicate request for method still in process");
                    }
         
                    return pjp.proceed();
                } catch (Exception e) {
                    throw e;
                } finally {
                    // 解锁
                    this.unLock(lockObj);
                }
            }
         
            /**
             * @param pjp
             * @return
             * @throws NoSuchMethodException
             */
            private DistributedLock getDistributedLock(ProceedingJoinPoint pjp) throws NoSuchMethodException {
                String methodName = pjp.getSignature().getName();
                Class clazz = pjp.getTarget().getClass();
                Class<?>[] par = ((MethodSignature) pjp.getSignature()).getParameterTypes();
                Method lockMethod = clazz.getMethod(methodName, par);
                DistributedLock distributedLock = lockMethod.getAnnotation(DistributedLock.class);
                return distributedLock;
            }
         
            /**
             * 解锁
             *
             * @param lockObj
             */
            private void unLock(ILock lockObj) {
                if (Objects.isNull(lockObj)) {
                    return;
                }
         
                try {
                    this.distributedLock.unLock(lockObj);
                } catch (Exception e) {
                    log.error("分布式锁解锁异常", e);
                }
            }
         
            /**
             * 获取 lockKey
             *
             * @param pjp
             * @param distributedLock
             * @return
             */
            private String getLockKey(ProceedingJoinPoint pjp, DistributedLock distributedLock) {
                String lockKey = distributedLock.key();
                String keyPrefix = distributedLock.keyPrefix();
                if (StringUtils.isBlank(lockKey)) {
                    throw new IdempotencyException("Lok key cannot be empty");
                }
                if (lockKey.contains("#")) {
                    this.checkSpEL(lockKey);
                    MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
                    // 获取方法参数值
                    Object[] args = pjp.getArgs();
                    lockKey = getValBySpEL(lockKey, methodSignature, args);
                }
                lockKey = StringUtils.isBlank(keyPrefix) ? lockKey : keyPrefix + lockKey;
                return lockKey;
            }
         
            /**
             * 解析spEL表达式
             *
             * @param spEL
             * @param methodSignature
             * @param args
             * @return
             */
            private String getValBySpEL(String spEL, MethodSignature methodSignature, Object[] args) {
                // 获取方法形参名数组
                String[] paramNames = nameDiscoverer.getParameterNames(methodSignature.getMethod());
                if (paramNames == null || paramNames.length < 1) {
                    throw new IdempotencyException("Lok key cannot be empty");
                }
                Expression expression = spelExpressionParser.parseExpression(spEL);
                // spring的表达式上下文对象
                EvaluationContext context = new StandardEvaluationContext();
                // 给上下文赋值
                for (int i = 0; i < args.length; i++) {
                    context.setVariable(paramNames[i], args[i]);
                }
                return expression.getValue(context).toString();
            }
         
            /**
             * SpEL 表达式校验
             *
             * @param spEL
             * @return
             */
            private void checkSpEL(String spEL) {
                try {
                    ExpressionParser parser = new SpelExpressionParser();
                    parser.parseExpression(spEL, new TemplateParserContext());
                } catch (Exception e) {
                    log.error("spEL表达式解析异常", e);
                    throw new IdempotencyException("Invalid SpEL expression [" + spEL + "]");
                }
            }
        }
3.  ### **定义分布式锁注解版启动元注解** ###
    
        @Target(ElementType.TYPE)
        @Retention(RetentionPolicy.RUNTIME)
        @Documented
        @Import({DistributedLockAspect.class})
        public @interface EnableDistributedLock {
        }
4.  ### 定义IDistributedLock分布式锁接口 ###
    
        public interface IDistributedLock {
            /**
             * 获取锁，默认30秒失效，失败一直等待直到获取锁
             *
             * @param key 锁的key
             * @return 锁对象
             */
            ILock lock(String key);
         
            /**
             * 获取锁,失败一直等待直到获取锁
             *
             * @param key      锁的key
             * @param lockTime 加锁的时间，超过这个时间后锁便自动解锁； 如果lockTime为-1，则保持锁定直到显式解锁
             * @param unit     {@code lockTime} 参数的时间单位
             * @param fair     是否公平锁
             * @return 锁对象
             */
            ILock lock(String key, long lockTime, TimeUnit unit, boolean fair);
         
            /**
             * 尝试获取锁，30秒获取不到超时异常，锁默认30秒失效
             *
             * @param key     锁的key
             * @param tryTime 获取锁的最大尝试时间
             * @return
             * @throws Exception
             */
            ILock tryLock(String key, long tryTime) throws Exception;
         
            /**
             * 尝试获取锁，获取不到超时异常
             *
             * @param key      锁的key
             * @param tryTime  获取锁的最大尝试时间
             * @param lockTime 加锁的时间
             * @param unit     {@code tryTime @code lockTime} 参数的时间单位
             * @param fair     是否公平锁
             * @return
             * @throws Exception
             */
            ILock tryLock(String key, long tryTime, long lockTime, TimeUnit unit, boolean fair) throws Exception;
         
            /**
             * 解锁
             *
             * @param lock
             * @throws Exception
             */
            void unLock(Object lock);
         
         
            /**
             * 释放锁
             *
             * @param lock
             * @throws Exception
             */
            default void unLock(ILock lock) {
                if (lock != null) {
                    unLock(lock.getLock());
                }
            }
         
         
        }
5.  ### IDistributedLock实现类 ###
    
        @Slf4j
        @Component
        public class RedissonDistributedLock implements IDistributedLock {
         
            @Resource
            private RedissonClient redissonClient;
            /**
             * 统一前缀
             */
            @Value("${redisson.lock.prefix:bi:distributed:lock}")
            private String prefix;
         
            @Override
            public ILock lock(String key) {
                return this.lock(key, 0L, TimeUnit.SECONDS, false);
            }
         
            @Override
            public ILock lock(String key, long lockTime, TimeUnit unit, boolean fair) {
                RLock lock = getLock(key, fair);
                // 获取锁,失败一直等待,直到获取锁,不支持自动续期
                if (lockTime > 0L) {
                    lock.lock(lockTime, unit);
                } else {
                    // 具有Watch Dog 自动延期机制 默认续30s 每隔30/3=10 秒续到30s
                    lock.lock();
                }
                return new ILock(lock, this);
            }
         
            @Override
            public ILock tryLock(String key, long tryTime) throws Exception {
                return this.tryLock(key, tryTime, 0L, TimeUnit.SECONDS, false);
            }
         
            @Override
            public ILock tryLock(String key, long tryTime, long lockTime, TimeUnit unit, boolean fair)
                    throws Exception {
                RLock lock = getLock(key, fair);
                boolean lockAcquired;
                // 尝试获取锁，获取不到超时异常,不支持自动续期
                if (lockTime > 0L) {
                    lockAcquired = lock.tryLock(tryTime, lockTime, unit);
                } else {
                    // 具有Watch Dog 自动延期机制 默认续30s 每隔30/3=10 秒续到30s
                    lockAcquired = lock.tryLock(tryTime, unit);
                }
                if (lockAcquired) {
                    return new ILock(lock, this);
                }
                return null;
            }
         
            /**
             * 获取锁
             *
             * @param key
             * @param fair
             * @return
             */
            private RLock getLock(String key, boolean fair) {
                RLock lock;
                String lockKey = prefix + ":" + key;
                if (fair) {
                    // 获取公平锁
                    lock = redissonClient.getFairLock(lockKey);
                } else {
                    // 获取普通锁
                    lock = redissonClient.getLock(lockKey);
                }
                return lock;
            }
         
            @Override
            public void unLock(Object lock) {
                if (!(lock instanceof RLock)) {
                    throw new IllegalArgumentException("Invalid lock object");
                }
                RLock rLock = (RLock) lock;
                if (rLock.isLocked()) {
                    try {
                        rLock.unlock();
                    } catch (IllegalMonitorStateException e) {
                        log.error("释放分布式锁异常", e);
                    }
                }
            }
        }
6.  ### 定义ILock锁对象 ###
    
        import lombok.AllArgsConstructor;
        import lombok.Getter;
         
        import java.util.Objects;
         
        /**
         * <p>
         * RedissonLock 包装的锁对象 实现AutoCloseable接口，在java7的try(with resource)语法，不用显示调用close方法
         * </p>
         * @since 2023-06-08 16:57
         */
        @AllArgsConstructor
        public class ILock implements AutoCloseable {
            /**
             * 持有的锁对象
             */
            @Getter
            private Object lock;
            /**
             * 分布式锁接口
             */
            @Getter
            private IDistributedLock distributedLock;
         
            @Override
            public void close() throws Exception {
                if(Objects.nonNull(lock)){
                    distributedLock.unLock(lock);
                }
            }
        }

## 六、使用示例 ##

**启动类添加@EnableDistributedLock启用注解支持**

@SpringBootApplication
    @EnableDistributedLock
    public class BiCenterGoodsApplication {
     
        public static void main(String[] args) {
            
            SpringApplication.run(BiCenterGoodsApplication.class, args);
            
        }
    }

**@DistributedLock标注需要使用分布式锁的方法**

@ApiOperation("编辑SKU供应商供货信息")
        
        @PostMapping("/editSupplierInfo")
        //@DistributedLock(key = "#dto.sku + '-' + #dto.skuId", lockTime = 10L, keyPrefix = "sku-")
        @DistributedLock(key = "#dto.sku", lockTime = 10L, keyPrefix = "sku-")
        public R<Boolean> editSupplierInfo(@RequestBody @Validated ProductSkuSupplierInfoDTO dto) {
            return R.ok(productSkuSupplierMeasureService.editSupplierInfo(dto));
        }

#dto.sku是 SpEL表达式。Spring中支持的它都支持的。比如调用静态方法，三目表达式。SpEL 可以使用方法中的任何参数。SpEL表达式参考

--------------------

[从原理到实践，分析 Redis 分布式锁的多种实现方案（一）][Redis]

[从原理到实践，分析 Redisson 分布式锁的实现方案（二）][Redisson]

[Spring Boot 集成 Redisson分布式锁][Spring Boot _ Redisson]

--------------------

参考文章：[一个注解，优雅的实现接口幂等性！][Link 1]

[Redis]: https://blog.csdn.net/Ascend1977/article/details/131221941?spm=1001.2014.3001.5502
[Redisson]: https://blog.csdn.net/Ascend1977/article/details/131373419?spm=1001.2014.3001.5502
[Spring Boot _ Redisson]: https://blog.csdn.net/Ascend1977/article/details/131126047?spm=1001.2014.3001.5502
[Link 1]: https://mp.weixin.qq.com/s/fqOB5YIDlaD8jjBpHzUuLQ