前言
同样是 面试问题
如何确保接口的 幂等性
幂等是一个 较为抽象的概念, 多次重复访问, 不会导致业务逻辑的异常
这里从增删改查, 几个方面列一下
一般来说, 我们核心需要关注的就是 新增 和 更新
对于 增加元素, 首先针对唯一约束进行校验, 然后再处理新增的相关业务, 严格一点需要 加锁, 分布式并发控制
对于 删除元素, 就是检查元素存不存在, 存在 则删除, 不存在 返回相关状态吗, 或者直接成功都 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;
}
}完
