【工作篇】接口幂等问题探究

前言

最近遇到一些问题，表单重复提交，导致插入重复数据到数据库，这里查询一些通用的方案，自己都实践一下，以后好回顾。

实践代码项目 Github： https://github.com/h-dj/Spring-Learning/tree/master/repeat-submit

一、什么是接口幂等性？

幂等含义

幂等（idempotent、idempotence）是一个数学与计算机学概念，常见于抽象代数中。

在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数。这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变。—- 百科

简单理解：就是针对一个操作，不管做多少次，产生的效果都是一样的。

举例：

  • 前端对同一表单数据的重复提交，后台应该只会产生一条记录
  • 我们发起一笔付款请求，应该只扣用户账户一次钱，当遇到网络重发或系统 bug 重发，也应该只扣一次钱
  • 发送消息，也应该只发一次，同样的短信如果多次发给用户，用户会崩溃
  • 创建业务订单，一次业务请求只能创建一个，不能出现创建多个订单

二、那些情况下会需要接口幂等性?

对于业务中需要考虑幂等性的地方一般都是接口的重复请求，重复请求是指同一个请求因为某些原因被多次提交。导致这个情况会有以下几种场景：

  • 前端重复提交：提交订单，用户快速重复点击多次，造成后端生成多个内容重复的订单。
  • 接口超时重试：对于给第三方调用的接口，为了防止网络抖动或其他原因造成请求丢失，这样的接口一般都会设计成超时重试多次。
  • 消息重复消费：MQ 消息中间件，消息重复消费。

三、接口幂等性解决方案

3.1、一般解决方案了解

前端方面：

  • 控制操作次数，例如：提交按钮仅可操作一次（提交动作后按钮置灰）
  • 及时重定向，例如：下单/支付成功后跳转到成功提示页面，这样消除了浏览器前进或后退造成的重复提交问题。

后台方面：

  • 数据库去重表(唯一索引)：
    • 利用数据库的唯一索引特性，保证唯一的逻辑
    • 使用唯一索引，可以是在原来的业务表添加唯一索引，也可以新建一个特定的去重表
  • 使用分布式锁：
    • 借助中间件：Redis 、Zookeeper
    • 注意要保证中间件的高可用
  • Token 机制：
    • 进入页面时先申请 Token，提交表单时再使用 Token 请求，后台根据 Token 判断是否重复请求
    • 前端申请 Token 和 携带 Token 的行为需要进行封装，不然每个页面都需要手动，比较繁琐，后台的话可以使用 注解+ AOP 统一拦截
  • 分布式项目，依然需要借助中间件：Redis

注意：

  • 一般推荐前端后台一起组合接口幂等方案，这样比较安全高效。

以下准备使用加入购物车为例，实现各个方案

3.2、数据库去重表(唯一索引)
  • 操作步骤
    • 1、通过请求的业务参数，组成唯一 ID
    • 2、通过 ID 查询去重表中是否存在记录，存在则抛出 重复请求异常（是否抛出异常，根据具体业务决定）
    • 3、否则，向去重表插入记录，如果插入异常，说明有多个请求同时执行，抛出重复请求异常
    • 4、去重表插入记录成功后，执行加入购物车操作
    • 5、执行加入购物车操作成功后，删除去重表记录
    • 注意： 去重表操作 和 业务处理要在同一个事物中，方式业务处理失败后，没有回滚去重表记录，导致商品加入购物车后，不能在增加数量
  • 创建表 
  1. DROP TABLE IF EXISTS `cart`;
  2. CREATE TABLE `cart` (
  3.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  4.   `product_id` bigint(20) DEFAULT NULL,
  5.   `product_name` varchar(500) DEFAULT NULL COMMENT '商品名称',
  6.   `product_sku_id` bigint(20) DEFAULT NULL,
  7.   `member_id` bigint(20) DEFAULT NULL,
  8.   `quantity` int(11) DEFAULT NULL COMMENT '购买数量',
  9.   `price` decimal(10,2) DEFAULT NULL COMMENT '添加到购物车的价格',
  10.   PRIMARY KEY (`id`)
  11. )
  12. ENGINE=InnoDB
  13. DEFAULT CHARSET=utf8mb4
  14. COLLATE=utf8mb4_0900_ai_ci
  15. COMMENT='购物车表';
  17. DROP TABLE IF EXISTS `deduplicate`;
  18. CREATE TABLE `deduplicate` (
  19.     id BIGINT(20) auto_increment NOT NULL,
  20.     unique_id varchar(100) NULL COMMENT '唯一ID',
  21.     CONSTRAINT deduplicate_PK PRIMARY KEY (id),
  22.     CONSTRAINT deduplicate_UN UNIQUE KEY (unique_id)  -- 唯一索引
  23. )
  24. ENGINE=InnoDB
  25. DEFAULT CHARSET=utf8mb4
  26. COLLATE=utf8mb4_0900_ai_ci
  27. COMMENT='去重表';
  • 添加到购物车逻辑
  1. @RestController
  2. @RequestMapping("/cart")
  3. public class CartController {
  4.     @Autowired
  5.     private ICartService iCartService;
  7.     @ApiRepeatUniqueIdSubmit(keyExpression = "@cartController.getUserId()+'_'+#cartPO.getProductId() +'_'+#cartPO.getProductSkuId()")
  8.     @PostMapping(value = "/add")
  9.     public String add(@RequestBody CartPO cartPO) throws InterruptedException {
  10.         cartPO.setMemberId(getUserId());
  11.         iCartService.addCart(cartPO);
  12.         return "ok";
  13.     }
  14.     /**
  15.      * 获取当前登录用户ID
  16.      *
  17.      * @return
  18.      */
  19.     public Long getUserId() {
  20.         return 1001L;
  21.     }
  22. }
  1. @Transactional(rollbackFor = Exception.class)
  2. @Override
  3. public void addCart(CartPO cartPO) throws InterruptedException {
  5.     LambdaQueryWrapper<CartPO> queryWrapper = Wrappers.<CartPO>lambdaQuery()
  6.         .eq(CartPO::getMemberId, cartPO.getMemberId())
  7.         .eq(CartPO::getProductId, cartPO.getProductId())
  8.         .eq(CartPO::getProductSkuId, cartPO.getProductSkuId());
  9.     //查询商品，已添加到购物车的，就增加数量即可(业务逻辑幂等)
  10.     //因为 select 和 save 操作不是串行执行的，可能有两个线程同时查询到商品没有添加到购物车
  11.     //然后同一个商品被两个线程分别入库了，导致购物车出现相同商品的两条记录
  12.     List<CartPO> list = this.list(queryWrapper);
  13.     //模拟耗时
  14.     TimeUnit.SECONDS.sleep(1);
  15.     if (list == null || list.isEmpty()) {
  16.         //添加到购物车
  17.         this.save(cartPO);
  18.     } else {
  19.         CartPO updateCartPO = list.get(0);
  20.         //数量加一
  21.         LambdaUpdateWrapper<CartPO> updateWrapper = Wrappers.<CartPO>lambdaUpdate()
  22.             .eq(CartPO::getId, updateCartPO.getId())
  23.             .setSql("quantity = quantity + 1");
  25.         this.update(updateWrapper);
  26.     }
  27. }
  • 编写 AOP 拦截处理
  1. @Target(ElementType.METHOD)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Documented
  4. @Inherited
  5. public @interface ApiRepeatUniqueIdSubmit {
  7.     /**
  8.      * 唯一key，支持Spring EL 表达式
  9.      *
  10.      * @return
  11.      * @ 符号引用 Spring 注册的bean
  12.      * # 符合引用方法上的参数
  13.      * param?.id  其中? 是避免param为空时，发生空指针异常
  14.      * @see <a>https://docs.spring.io/spring-framework/docs/3.0.x/reference/expressions.html</a>
  15.      */
  16.     String keyExpression();
  17. }
  1. @Component
  2. @Aspect
  3. public class ApiRepeatSubmitUniqueIdAspect {
  5.     @Autowired
  6.     private ApplicationContext applicationContext;
  8.     @Autowired
  9.     private IDeduplicateService iDeduplicateService;
  11.     @Pointcut("@annotation(cn.hdj.repeatsubmit.aspect.ApiRepeatUniqueIdSubmit)")
  12.     public void pointCut() {
  13.     }
  14.     @Transactional(rollbackFor = Exception.class)
  15.     @Around("pointCut()")
  16.     public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
  18.         Signature signature = joinPoint.getSignature();
  19.         MethodSignature msig = (MethodSignature) signature;
  20.         Method targetMethod = msig.getMethod();
  21.         ApiRepeatUniqueIdSubmit apiRepeatSubmit = targetMethod.getAnnotation(ApiRepeatUniqueIdSubmit.class);
  22.         String keyExpression = apiRepeatSubmit.keyExpression();
  24.         Map<String, Object> argMap = SpringElUtil.getArgMap(joinPoint);
  25.         //获取业务参数，组成唯一ID
  26.         String uniqueId = SpringElUtil.<String>createElBuilder()
  27.                 .setArgMap(argMap)
  28.                 .setBeanFactory(applicationContext)
  29.                 .setTarget(String.class)
  30.                 .setKeyExpression(keyExpression)
  31.                 .build();
  33.         LambdaQueryWrapper<DeduplicatePO> queryWrapper = Wrappers.<DeduplicatePO>lambdaQuery()
  34.                 .eq(DeduplicatePO::getUniqueId, uniqueId);
  36.         long count = this.iDeduplicateService.count(queryWrapper);
  37.         if (count > 0) {
  38.             throw new DuplicateKeyException("不要重复提交!");
  39.         }
  41.         //插入去重表
  42.         DeduplicatePO deduplicatePO = new DeduplicatePO();
  43.         deduplicatePO.setUniqueId(uniqueId);
  44.         try {
  45.             this.iDeduplicateService.save(deduplicatePO);
  46.         } catch (Exception e) {
  47.             throw new DuplicateKeyException("不要重复提交!");
  48.         }
  50.         Object proceed = joinPoint.proceed();
  52.         //执行完删除
  53.         this.iDeduplicateService.removeById(deduplicatePO);
  55.         return proceed;
  56.     }
  57. }

3.3、分布式锁

分布式锁可以使用 Redis 和 Zookeeper ，更多关于 Redis 和 Zookeeper 的使用 请自行查阅资料。以下使用 Redis 来实现分布式锁

  • 引入依赖
  1. <dependency>
  2.     <groupId>org.redisson</groupId>
  3.     <artifactId>redisson-spring-boot-starter</artifactId>
  4.     <version>3.16.6</version>
  5. </dependency>
  • 配置
  1. spring:
  2.   redis:
  3.     # https://github.com/redisson/redisson/wiki/2.-Configuration
  4.     database: '0'
  5.     host: '127.0.0.1'
  6.     port: '6379'
  7.     #password: '123456'
  8. #    ssl:
  9. #    timeout:
  10. #    cluster:
  11. #      nodes:
  12. #    sentinel:
  13. #      master:
  14. #      nodes:
  • 在 AOP 获取锁的逻辑
  1. //创建锁
  2. RLock lock = this.redissonClient.getLock(LOCK_PREFIX + uniqueId);
  4. //判断是否被抢占了锁
  5. if (lock.isLocked()) {
  6.     throw new DuplicateKeyException("不要重复提交!");
  7. }
  9. //尝试获取锁， 默认30秒会超时过期， 并启动线程监听，自动续签
  10. //当客户端异常，终止了续签线程，超时后会删除锁，避免发生死锁
  11. //如果自己手动设置了超时过期时间，则不会启动线程监听，自动续签
  12. if (lock.tryLock()) {
  13.     try {
  14.         return joinPoint.proceed();
  15.     } finally {
  16.         //释放锁
  17.         lock.unlock();
  18.     }
  19. }
  20. throw new DuplicateKeyException("不要重复提交!");

3.4、使用 Token 机制
  • 操作步骤
    • 1、在访问页面时，先获取 Token ，保持到指定的地方
    • 2、在点击加入购物车时，把 Token 放到 Header 或请求参数中，带给后台
    • 3、后台验证 Token 并删除，表示该 Token 已使用
    • 4、执行加入购物车逻辑
    • 5、成功响应前端（看业务需求，是否需要重新获取 Token）
  • 创建 Token 生成和验证服务 
  1. @Service
  2. public class TokenServiceImpl implements TokenService {
  4.     public static final String TOKEN_PREFIX = "token_idempotent";
  5.     public static final String TOKEN_HEADER_NAME = "x-token";
  6.     public static final Long TOKEN_EXPIRE_TIME = 5 * 60L;
  8.     @Autowired
  9.     private RedissonClient redissonClient;
  11.     @Override
  12.     public String createToken() {
  13.         String ID = UUID.randomUUID().toString();
  14.         RBucket<String> bucket = this.redissonClient.<String>getBucket(String.format("%s:%s", TOKEN_PREFIX, ID), StringCodec.INSTANCE);
  15.         //默认超时5分钟
  16.         bucket.trySet(ID, TOKEN_EXPIRE_TIME, TimeUnit.SECONDS);
  17.         return ID;
  18.     }
  19.     @Override
  20.     public boolean checkToken(HttpServletRequest request) {
  21.         // 从请求头中获取token信息
  22.         String token = request.getHeader(TOKEN_HEADER_NAME);
  23.         if (!StringUtils.hasLength(token)) {
  24.             //从参数中获取token值
  25.             token = request.getParameter(TOKEN_HEADER_NAME);
  26.         }
  27.         if (!StringUtils.hasLength(token)) {
  28.             throw new DuplicateKeyException("重复提交，提交失败");
  29.         }
  31.         RBucket<String> bucket = this.redissonClient.getBucket(String.format("%s:%s", TOKEN_PREFIX, token), StringCodec.INSTANCE);
  33.         //获取，并删除
  34.         String ID = bucket.getAndDelete();
  35.         //不存在，则重复提交
  36.         if (ID == null) {
  37.             throw new DuplicateKeyException("重复提交，提交失败");
  38.         }
  39.         return true;
  40.     }
  41. }
  • 在 AOP 中检查
  1. @Component
  2. @Aspect
  3. public class ApiRepeatSubmitTokenAspect {
  4.     @Autowired
  5.     private TokenService tokenService;
  7.     @Autowired
  8.     private HttpServletRequest request;
  10.     @Pointcut("@annotation(cn.hdj.repeatsubmit.aspect.ApiRepeatTokenSubmit)")
  11.     public void pointCut() {
  12.     }
  13.     @Before("pointCut()")
  14.     public void Before(JoinPoint joinPoint) {
  15.         tokenService.checkToken(request);
  16.     }
  17. }
  • 前端进入页面时是先获取 Token
  1. $.ajax({
  2.     type: "GET",
  3.     url: "/token/create",
  4.     contentType:'application/json',
  5.     success: function(token){
  6.         sessionStorage.setItem('x-token',token);
  7.     }
  8. });
  • 请求时带上 Token
  1. $("#addCart").click(function(){
  2.     //按钮置灰
  3.     $("#addCart").attr('disabled','disabled');
  4.     $.ajax({
  5.         type: "POST",
  6.         url: "/cart/add",
  7.         headers:{
  8.             //携带token
  9.             'x-token':sessionStorage.getItem('x-token')
  10.         },
  11.         data: JSON.stringify({
  12.             productId: $('input[name=product_id]').val(),
  13.             productSkuId: $('input[name=product_sku_id]').val(),
  14.             productName: $('input[name=product_name]').val(),
  15.             price: $('input[name=price]').val(),
  16.             quantity: $('input[name=quantity]').val()
  17.         }),
  18.         contentType:'application/json',
  19.         success: function(msg){
  20.             console.log(msg)
  21.         },
  22.         complete: function(msg){
  23.             $("#addCart").removeAttr('disabled');
  24.         }
  25.     });
  26. });

总结

以上是较为常见通用的幂等方案，但实际业务可能比较个性化，需要跟业务结合进行考虑，采用合适的方法或结合使用，例如：

  • 如果该业务是存在状态流转，可以采用状态机策略进行业务幂等判断
  • 如果该业务是更新数据，可以采用多版本策略，在需要更新的业务表上加上版本号

参考