batchsize一定是2的幂_分布式接口幂等性方案的讨论（自定义注解实现接口幂等）...

亦凉 2023-01-01 10:48 202阅读 0赞

*  2.1、token机制
 *  2.2、各种锁机制
    
     *  1、数据库悲观锁
     *  2、数据库乐观锁
     *  3、各种唯一约束
 *  3.1、注解方式实现接口幂等性
 *  3.2、使用自定义注解实现接口自动幂等

**前言**

在多服务的情况下，有的接口是需要保证幂等的。比如扣款接口与订单接口的两次调用，如果不保证接口的幂等。那么该接口很有可能会出现网络延迟等问题，扣款在订单之前进行，那么在订单反应过来后，又调用了一次扣款。这样就扣款两次。第二天就不用来上班了！

# **1.什么情况下需要幂等** #

*  以SQL为例，有些操作是天然幂等的。
    
     *      SELECT * FROM table WHERE id = ?
            - 无论执行多少次都不会改变状态，是天然幂等的
     *      UPDATE table SET col1 = 1 HWERE col2 = 2
            - 无论执行多少次都不会改变状态，是天然幂等的
     *      delete from user where userid = 1
            - 多次操作结果一样，具备幂等性
     *      INSERT into user(userid,name) values(1,'a')
            - 如userid为唯一主键，即重复操作上面的业务，也只会插入一条数据，具备幂等
 *  每次更新都会影响执行结果的操作都是不具备幂等的
    
     *      update tab1 set col1 = col1 + 1 where col2 =2 
            - 每次执行的结果都会发生变化，不具备幂等
     *      INSERT into user(userid,name) values(1,'a')
            - 如userid不是唯一主键，可以重复，即重复操作上面的业务，也只会插入一条数据，不具备幂等

# **2、解决幂等的几种方案** #

## **2.1、token机制** ##

1.  在服务端提供了发送token的接口，在分析业务的时候，哪些业务是存在幂等问题的。就必须在执行业务前，先去获取token，服务器会把token保存到redis中。
2.  然后调用业务接口请求时，吧token且带过去，一般放在请求头部。
3.  服务器判断token是否存在redis中，存在表示第一次请求，然后删除token，继续执行业务。
4.  如果判断token不存在redis中，就表示重复操作，直接返回重复标记给client，这样就保证了业务代码，不被重复执行。

**危险性**

1.  先删除token还是后删除token
    
    > 最好的解决方案是，先删除token，如果业务调用失败，就重新获取token再次请求。
    
    1.  先删除可能导致，业务确实没有执行，重试还带上之前token，可能有些公司业务里面做了防重设计导致请求还是不能执行。
    2.  后删除可能导致，业务处理成功，但是服务闪断，出现超时，没有删除token，别人继续重试，导致业务被执行两遍。
2.  Token获取比较和删除必须是原子性的
    
    >     if redis.call('get',KEY[1])==ARGV[1]then return redis.call('del',KEY[1])else return 0 end
    
    1.  redis.get(token),token.equals,redis.del(token)。如果这两个操作不是原子，可能导致，高并发下，都get到同样的数据，判断成功，继续业务并发执行
    2.  可以在redis使用lua脚本完成这一套操作(在redis中 lua脚本的执行是保证原子性的)

## **2.2、各种锁机制** ##

### **1、数据库悲观锁** ###

select * from xxxx where id = 1 for update;

悲观锁使用时一般伴随事务一起使用，数据锁定时间可能很长，需要根据实际情况选用。另外要注意的是，id字段一定是主键或者唯一索引，不然可能造成**锁表**的结果，处理起来会非常麻烦。

### **2、数据库乐观锁** ###

update t_goods set count =count-1,version = version + 1 where good_id=2 and version = 1

根据version版本，也就是在操作数据库前先获取当前字段的version版本号，然后操作的时候带上此version号。

第一次操作库存是，得到version为1，调用库存服务version变成了2，但返回给订单服务出现了问题，订单服务又一次发起调用库存服务，当订单服务传入的version还是1，再执行上面的sql语句时，就不会执行；因为version已经变成2了，where条件就不成立了。这样就保证了不管调用几次，只会真正的处理一次。

乐观锁主要适用于处理读多写少的问题。

### **3、各种唯一约束** ###

1.  **数据库唯一约束**
    
    1.  插入数据，应该按照唯一索引进行插入，比如订单号，相同的订单就不可能有两条记录插入。
    2.  在数据库层面防止重复
    3.  这个机制是利用了数据库的主键唯一约束的特性，解决了在insert常见时幂等问题，单主键的要求不是自增的主键，这样就需要业务生成全局唯一的主键。
    4.  如果是分库分表场景下，路由规则要保证相同请求下，落地在同一个数据库和同一表中，要不然数据库主键约束就不起效果了，因为是不同的数据库和表主键不相关
2.  **redis set防重**
    
    1.  很多数据需要处理，只能被处理一次，比如将计算的数据MD5将其放入redis的set，每次处理数据，先看这个MD5是否已经存在，存在就不处理。
3.  **业务层分布式锁**
    
    1.  如果多个机器可能在同一时间同时处理相同的数据，比如多台机器定时任务都拿到了相同数据处理，我们就可以加分不是锁，锁定此数据，处理完后释放锁。获取到锁的必须先判断这个数据是否被处理过。
4.  **防重表**
    
    1.  有点类似于redis 的set。
    2.  比如说订单表中，使用订单号orderCn作为去重表的唯一索引，吧唯一索引插入去重表，再进行业务操作，他们在同一个事务中，这个保证了重复请求时，因为去重表有唯一约束，导致请求失败，避免了幂等问题。这里要注意的是，去重表和业务表应该在同一库中，这样就保证了在同一个事务中，即使业务操作失败了，也会把去重表的数据回滚，这个很好的保证了数据的一致性。
5.  **全局请求唯一id**
    
        prox_set_header X-Request-id $request_id;
    
    1.  调用接口是，生成一个唯一id，redis将数据保存到集合中(去重)，存在及处理过。可以使用nginx设置每一个请求的唯一id；

\#3、实际业务的幂等处理

在我们公司所技术调研时，关于接口幂等处理使用的是上述的第五种情况。全局请求的唯一id加上redis的去重表。但是在需要幂等处理的接口生成全局唯一id

既然是学习，就对公司的部分代码进行重构，实现一个简单的基于注解接口幂等解决方案

## **3.1、注解方式实现接口幂等性** ##

![f03cc702f7befdcbf3a50492e1f2977b.png][] image-20201015123503280

> 上面是一个简单的接口幂等的业务流程。
> 
> 使用到的技术栈 spring boot + redis + 加上雪花算法的全局唯一id

## **3.2、使用自定义注解实现接口自动幂等** ##

效果图：

![2e708fb515dfbc1d7904f8db8ca4996c.png][] image-20201210205908132

主要逻辑：

/**
     * AOP 实现接口幂等注解的一键幂等
     *
     * @author by Mr. Li 2020/12/10 17:27
     */
    @Aspect
    @Component
    public class ApiIdempotentAspect {
         
    
    /**
         * 实现 redis 的 get delete 原子性 的lua 脚本
         */
    private static final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private IdWorker idWorker;
    
    
    /**
         * 切面
         */
    @Pointcut("@annotation(com.lg.distributed.idempotence.annotation.AutoIdempotent)")
    public void pointCut() {
         
        }
    
    @Around("pointCut()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
         
    // 获取 使用注解的方法
            MethodSignature signature = (MethodSignature) joinPoint.getSignature();
            Method method = signature.getMethod();
            AutoIdempotent autoIdempotent = method.getDeclaredAnnotation(AutoIdempotent.class);
    if (autoIdempotent != null) {
         
    return apiIdempotent(joinPoint, signature, method, autoIdempotent);
            }
    // 放行
            Object proceed = joinPoint.proceed();
    return proceed;
    
        }
    
    /**
         * 校验token
         *
         * @param joinPoint
         * @param signature
         * @param method
         * @param autoIdempotent
         * @return
         * @throws Throwable
         */
    
    private Object apiIdempotent(ProceedingJoinPoint joinPoint, MethodSignature signature, Method method, AutoIdempotent autoIdempotent) throws Throwable {
         
    if (autoIdempotent == null) {
         
    // 直接放行
    return joinPoint.proceed();
            }
    // 1. 生成 全局唯一id
            String uniqueId = String.valueOf(idWorker.nextId());
    // 当前接口就对应着当前id
            String prefix = autoIdempotent.prefix();
    if (StringUtils.isBlank(prefix)) {
         
    throw new GlobalException("CacheLock prefix can't be null");
            }
    // 拼接 key
            String delimiter = autoIdempotent.delimiter();
    
            StringBuilder sb = new StringBuilder();
    final String lockKey = sb.append(prefix).append(delimiter).append(method.getName()).toString();
    
    try {
         
    // 加锁
                Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, uniqueId, autoIdempotent.expire(), autoIdempotent.timeUnit());
    if (!result) {
         
    // 设置失败
    throw new GlobalException("请勿重复提交");
                }
    return joinPoint.proceed();
            } finally {
         
    // 释放锁。
                DefaultRedisScript redisScript = new DefaultRedisScript<>(RELEASE_LOCK_LUA_SCRIPT, Long.class);
                Long result = (Long) redisTemplate.execute(redisScript, Collections.singletonList(lockKey), uniqueId);
            }
        }// 这里提供另一种方案。/*
         * 假设，在nginx 的请求中设置了，id标识。那么在此可以进行请求拦截。实现请求携带token。
         * 并且如果该token是有序的。还可以配合队列 实现有序的网络请求
         * */public HttpServletRequest getRequest() {
         
            ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
            HttpServletRequest request = attributes.getRequest();return request;
        }
    }

>  *  以上实现逻辑有点类似使用了redis的分布式锁的样子。
>  *  反正逻辑都差不多吧
> 
> https://gitee.com/ligangyun/wheel/tree/master/distributed-idempotence
> 
> 代码

[f03cc702f7befdcbf3a50492e1f2977b.png]: /images/20221120/b3eb134f99f44cae8fdabd58db1cff8e.png
[2e708fb515dfbc1d7904f8db8ca4996c.png]: /images/20221120/c694e23c2dad46d485860e74dfaa434a.png