前言

同样是 面试问题 

如何确保接口的 幂等性 

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

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

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

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

对于 删除元素, 就是检查元素存不存在, 存在 则删除, 不存在 返回相关状态吗, 或者直接成功都 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();// lockif (userRunningStore.contains(name)) {throw new RuntimeException(String.format("有其他用户在新增用户 %s, 请刷新后重试", name));}userRunningStore.add(name);// unlockObject result = point.proceed();return result;}}

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

@PutMapping("/user")
@ConcurrentLatch
public AuthUser add(AuthUser user) {// physic verifyif (user.getAge() > 0 && user.getAge() < 111) {throw new RuntimeException("用户的 age 必须在合法的区间");}// logistic verifyMap<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 insertexistsUser = 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 limitif (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";// lockif (!IdempotentController.tokenStore.contains(token)) {throw new RuntimeException("服务器异常, 请刷新后重试");}IdempotentController.tokenStore.remove(token);// unlockObject 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 {@Resourceprivate JdbcTemplate jdbcTemplate;@PostMapping("/user")@ConcurrentLatchpublic AuthUser update(AuthUser user) {// physic verifyif (user.getAge() > 0 && user.getAge() < 111) {throw new RuntimeException("用户的 age 必须在合法的区间");}// logistic verifyMap<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 insertString 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;}}

完