114 接口中幂等性的保证

前言

同样是 面试问题 

如何确保接口的 幂等性 

幂等是一个 较为抽象的概念, 多次重复访问, 不会导致业务逻辑的异常 

这里从增删改查, 几个方面列一下 

一般来说, 我们核心需要关注的就是 新增 和 更新

对于 增加元素, 首先针对唯一约束进行校验, 然后再处理新增的相关业务, 严格一点需要 加锁, 分布式并发控制 

对于 删除元素, 就是检查元素存不存在, 存在 则删除, 不存在 返回相关状态吗, 或者直接成功都 OK

元素的新增

基于持久化的数据库的机制

比如 mysql 这边目标表, 增加唯一索引, 或者 主键

比如, 我们这里 限定在 用户表 中 用户名 不能重复, 这个只有特定的业务场景中可以这么处理 

CREATE TABLE `auth_user` (
  `id` int(11) NOT NULL,
  `name` varchar(256) DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `name` (`name`) USING BTREE COMMENT 'name'
) ENGINE=InnoDB DEFAULT CHARSET=utf8

然后 服务器这边 就不用做 过多的控制, 核心业务部分直接 ”insert into” 都可以 

由 mysql 这边本身的机制 来确保 用户名 的不能重复, 防止 用户多次提交 造成的业务问题

分布式并发过滤控制 + 数据库的悲观锁

我们这里展现一下 完整的处理流程, 主要是包含了 外层的并发过滤控制, 数据库校验控制, 数据库加锁+插入 控制

这里分布式并发控制这里模拟实现, 是 userRunningStore 部分

1. 并发过滤控制这边处理如下, 基于 spring 的 注解 + aop

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface ConcurrentLatch {

}

并发过滤控制的处理如下

@Component
@Aspect
public class ConcurrentLatchAop {

    @Pointcut("@annotation(ConcurrentLatch)")
    public void concurrentLatchAop() {

    }

    Set<String> userRunningStore = new LinkedHashSet<>();

    @Around("concurrentLatchAop()")
    public Object doProcess(ProceedingJoinPoint point) throws Throwable {
        Object[] args = point.getArgs();
        AuthUser user = (AuthUser) args[0];

        String name = user.getName();
        // lock
        if (userRunningStore.contains(name)) {
            throw new RuntimeException(String.format("有其他用户在新增用户 %s, 请刷新后重试", name));
        }
        userRunningStore.add(name);
        // unlock

        Object result = point.proceed();
        return result;
    }

}

数据库校验控制, 数据库加锁+插入控制 如下 

@PutMapping("/user")
@ConcurrentLatch
public AuthUser add(AuthUser user) {
    // physic verify
    if (user.getAge() > 0 && user.getAge() < 111) {
        throw new RuntimeException("用户的 age 必须在合法的区间");
    }
    // logistic verify
    Map<String, Object> existsUser = JdbcTemplateUtils.queryOne(jdbcTemplate, String.format(" select * from auth_user where name = '%s'; ", user.getName()));
    if (existsUser != null) {
        throw new RuntimeException("该用户已经存在, 用户名称不能重复");
    }

    // do other biz

    // lock then insert
    existsUser = JdbcTemplateUtils.queryOne(jdbcTemplate, String.format(" select * from auth_user where name = '%s' for update; ", user.getName()));
    if (existsUser != null) {
        throw new RuntimeException("该用户已经存在, 用户名称不能重复");
    }
    jdbcTemplate.execute(String.format("INSERT INTO `auth_user`(`name`, `age`) VALUES ('%s', %s);", user.getName(), user.getAge()));
    return user;
}

token分布式并发控制 + 数据库的悲观锁

这个就主要是 整体的交互机制调整, 增加了一层 token 的获取 和 验证

token 的分派这边如下, 做限流, 生成 token 的相关处理 

public static Map<String, AtomicInteger> interf2Counter = new LinkedHashMap<>();
public static Set<String> tokenStore = new LinkedHashSet<>();

// pre install all interfs
static {
    interf2Counter.put("IdempotentController.add", new AtomicInteger());
}

@GetMapping("/requestToken")
public String requestToken(String interf) {
    AtomicInteger counter = interf2Counter.get(interf);
    int incred = counter.getAndIncrement();
    // rate limit
    if (incred > 20) {
        counter.getAndDecrement();
        throw new RuntimeException(" 服务器繁忙, 请稍后重试 ");
    }

    String token = UUID.randomUUID().toString();
    String compositeToken = interf + token;
    tokenStore.add(compositeToken);
    return token;
}

并发控制这边处理如下 

/**
 * ConcurrentLatchAop
 *
 * @author Jerry.X.He
 * @version 1.0
 * @date 2023/9/21 10:17
 */
@Component
@Aspect
public class ConcurrentLatchAop {

    @Pointcut("@annotation(ConcurrentLatch)")
    public void concurrentLatchAop() {

    }

    @Around("concurrentLatchAop()")
    public Object doProcess(ProceedingJoinPoint point) throws Throwable {
        Object[] args = point.getArgs();
        String interf = "get interf from request";
        String token = "get token from request";

        // lock
        if (!IdempotentController.tokenStore.contains(token)) {
            throw new RuntimeException("服务器异常, 请刷新后重试");
        }
        IdempotentController.tokenStore.remove(token);
        // unlock

        Object result = point.proceed();
        AtomicInteger counter = IdempotentController.interf2Counter.get(interf);
        counter.getAndDecrement();
        return result;
    }

}

元素的更新

以上 三种处理方式 在元素的更新中同样可以使用

元素的更新 数据库的更新控制这边可以使用 基于数据库的乐观锁 

数据库的乐观锁更新

并发控制这边 和上面类似, 我们这里着重关注 数据库的更新这边 

数据库的更新这边, 主要是增加一个版本号的字段, 然后 更新的时候 在原有的 id 条件之外, 再增加一个 version 控制的字段 

根据 mysql 这边更新, 会增加行排他锁, 具体的处理如下 

/**
 * IdempotentController
 *
 * @author Jerry.X.He
 * @version 1.0
 * @date 2023/9/21 9:58
 */
@RestController
@RequestMapping("/idempotent")
public class IdempotentController {

    @Resource
    private JdbcTemplate jdbcTemplate;

    @PostMapping("/user")
    @ConcurrentLatch
    public AuthUser update(AuthUser user) {
        // physic verify
        if (user.getAge() > 0 && user.getAge() < 111) {
            throw new RuntimeException("用户的 age 必须在合法的区间");
        }
        // logistic verify
        Map<String, Object> existsUser = JdbcTemplateUtils.queryOne(jdbcTemplate, String.format(" select * from auth_user where name = '%s'; ", user.getName()));
        if (existsUser == null) {
            throw new RuntimeException("该用户不存在, 请确认输入");
        }

        // do other biz

        // lock then insert
        String id = String.valueOf(existsUser.get("id"));
        String version = String.valueOf(existsUser.get("version"));
        int updatedCount = jdbcTemplate.update(String.format("update auth_user set name = '%s', age = %s where id = %s and version = %s;", user.getName(), user.getAge(), id, version));
        if (updatedCount == 0) {
            throw new RuntimeException("该用户信息已经发生改变, 请刷新后重试");
        }
        return user;
    }

}

完