IT评测·应用市场-qidao123.com
标题:
业务幂等性技术架构体系之服务幂等深入分析
[打印本页]
作者:
吴旭华
时间:
2025-1-16 16:47
标题:
业务幂等性技术架构体系之服务幂等深入分析
简单来说，
服务幂等
是指一个操作（或哀求）无论被执行多少次，其对体系状态的影响都是一样的，就似乎这个操作只执行了一次一样。换句话说，对于同一组输入参数，幂等的服务会给出相同的结果，并且不会改变体系的最终状态。例如：
添加商品到购物车
：假如用户多次点击“参加购物车”按钮，应该只会增加一件商品，而不是每次都新增一件。
支付订单
：当用户尝试支付同一个订单时，岂论支付哀求发送了多少次，体系都应确保该订单只会被支付一次。
与接口幂等的区别
特征服务幂等（Service Idempotency）接口幂等（API Idempotency）
作用范围
整个业务流程，大概跨越多个API接口单个API接口，专注于技术实现
依赖因素
详细业务逻辑，如订单处置惩罚、库存管理等HTTP方法选择，API筹划原则
长期化影响
涉及数据库或其他长期存储的状态更新紧张是API相应行为，不肯定涉及长期化状态
示例
支付服务、注册服务GET/PUT/DELETE哀求 接下来我们将分别介绍几种服务幂等的办理方案
一、防重表
对于防止数据重复提交，还有一种办理方案就是通过防重表实现。防重表的实现思路也非常简单。首先创建一张表 作为防重表，同时在该表中创建一个或多个字段的唯一索引作为防重字段，用于保证并发情况下，数据只有一条。 在向业务表中插入数据之前先向防重表插入，假如插入失败则表示是重复数据。
对于防重表的办理方案，大概有人会说为什么不使用灰心锁。灰心锁在使用的过程中也是会发生死锁的。灰心锁是 通过锁表的方式实现的。 假设如今一个用户A访问表A（锁住了表A），然后试图访问表B； 另一个用户B访问表 B（锁住了表B），然后试图访问表A。 这时对于用户A来说，由于表B已经被用户B锁住了，所以用户A必须比及用 户B释放表B才能访问。 同时对于用户B来说，由于表A已经被用户A锁住了，所以用户B必须比及用户A释放表A才 能访问。此时死锁就已经产生了。
二、select+insert防重提交
说白了，就是在插入之前，先查询一下数据库是否存在，存在了，则不插入，不存在再新增插入
对于一些后台体系，并发量并不高的情况下，对于幂等的实现非常简单，通过select+insert思想即可完成幂等控制。
详细的，我们的业务流程总结为下图：
代码实现如下：
@Override
@Transactional(rollbackFor = Exception.class)
public String addOrder(Order order) {
order.setCreateTime(new Date());
order.setUpdateTime(new Date());
//查询
Order orderResult = orderMapper.selectByPrimaryKey(order.getId());
Optional<Order> orderOptional = Optional.ofNullable(orderResult);
if (orderOptional.isPresent()){
return "repeat request";
}
int result = orderMapper.insert(order);
if (result != 1){
return "fail";
}
return "success";
}
复制代码
但是我们颠末Jemeter 压测模仿100个线程，发现并不能保证幂等性
为什么呢？
我们分析，由于代码中，我们是先查询，再插入，这是两步操作（并非原子性操作），那么在这个间隙中，就会产生题目无法保证幂等性
办理方案：可以在方法上加上synchronized锁，但是这就降低体系承载的并发能力了
三、MySQL乐观锁
假设如今订单已经生成成功，那么就会涉及到扣减库存的操作。当高并发下同时扣减库存时，非常容易出现数据错误题目。
我们假如直接使用上面这段代码，会造成大量的数据不一致线程安全题目
但是假如我们加上synchronized呢?
我们用Jemeter模仿10000个线程扣减初始库存量为100000的库存，还是有题目
@Service
public class StockServiceImpl implements StockService {
@Autowired
private StockMapper stockMapper;
@Override
@Transactional(rollbackFor = Exception.class)
public synchronized int lessInventory(String goodsId, int num) {
return stockMapper.lessInventory(goodsId, num);
}
}
复制代码
当前已经在在方法上添加了synchronized，对当前方法对象举行了锁定。 通过Jemeter，模仿一万并发对其举行 访问。可以发现，仍然出现了脏数据。
该题目的产生缘故原由，就在于在方法上synchronized搭配使用了@Transactional。首先synchronized锁定的是当 前方法对象，而@Transactional会对当前方法举行AOP增强，动态代理出一个代理对象，在方法执行前开启事 务，执行后提交事件。 所以synchronized和@Transactional其实操作的是两个不同的对象，换句话说就是 @Transactional的事件操作并不在synchronized锁定范围之内。
假设A线程执行完扣减库存方法，会释放锁并提交事件。但A线程释放锁但还没提交事件前，B线程执行扣减库存方 法，B线程执行后，和A线程一起提交事件，就出现了线程安全题目，造成脏数据的出现。
那我们怎样通过MySQL乐观锁保证幂等呢，请看下文分析
MySQL乐观锁保证幂等
MySQL乐观锁是基于数据库完成分布式锁的一种实现，实现的方式有两种：基于版本号、基于条件。但是实现思想都是基于MySQL的行锁思想来实现的。
基于版本号实现
1）修改数据表，添加version字段，默认值为0
2）修改StockMapper添加基于版本修改数据方法
@Update("update tb_stock set amount=amount‐#{num},version=version+1 where goods_id=#{goodsId} and version=#{version}")
int lessInventoryByVersion(@Param("goodsId") String goodsId,@Param("num") int num,@Param("version") int version);
复制代码
3）测试模仿一万并发举行数据修改，此时可以发现当前版本号从0变为1，且库存量精确。
基于条件实现
通过版本号控制是一种非常常见的方式，适合于大多数场景。但如今库存扣减的场景来说，通过版本号控制就是多人并发访问购买时，查询时表现可以购买，但最终只有一个人能成功，这也是不可以的。其实最终只要商品库存不 发生超卖就可以。那此时就可以通过条件来举行控制。
1）修改StockMapper：
@Update("update tb_stock set amount=amount‐#{num} where goods_id=#{goodsId} and amount‐# {num}>=0")
int lessInventoryByVersionOut(@Param("goodsId") String goodsId,@Param("num") int num);
复制代码
2）修改StockController：
@PutMapping("/lessInventoryByVersionOut/{goodsId}/{num}")
public String lessInventoryByVersionOut(@PathVariable("goodsId") String goodsId,
@PathVariable("num") Integer num) throws InterruptedException {
System.out.println("reduce stock");
int result = stockService.reduceStockNoLock(goodsId, num);
if (result != 1){
return "reduce stock fail";
}
//延迟
TimeUnit.SECONDS.sleep(6000);
return "reduce stock success";
}
复制代码
3）通过jemeter举行测试，可以发现当多人并发扣减库存时，控制住了商品超卖的题目
四、zookeeper分布式锁
实现原理
对于分布式锁的实现，zookeeper天然携带的一些特性能够很完美的实现分布式锁。其内部紧张是使用znode节点 特性和watch机制完成。
在zookeeper中节点会分为四类，分别是：
长期节点：一旦创建，则永久存在于zookeeper中，除非手动删除。
长期有序节点：一旦创建，则永久存在于zookeeper中，除非手动删除。同时每个节点都会默认存在节点序 号，每个节点的序号都是有序递增的。如demo01、demo02.....demo0N。
暂时节点：当节点创建后，一旦服务器重启或宕机，则被主动删除。
暂时有序节点：当节点创建后，一旦服务器重启或宕机，则被主动删除。同时每个节点都会默认存在节点序 号，每个节点的序号都是有序递增的。如demo01、demo002.....demo00N。
watch监听机制紧张用于监听节点状态变更，用于后续事件触发，假设当B节点监听A节点时，一旦A节点发生修 改、删除、子节点列表发生变更等事件，B节点则会收到A节点改变的通知，接着完成其他额外事情。
实现原理如下：
其实现思想是当某个线程要对方法加锁时，首先会在zookeeper中创建一个与当前方法对应的父节点，接着每个要 获取当前方法的锁的线程，都会在父节点下创建一个暂时有序节点，由于节点序号是递增的，所以后续要获取锁的 线程在zookeeper中的序号也是逐次递增的。根据这个特性，当前序号最小的节点肯定是首先要获取锁的线程，因 此可以规定序号最小的节点获得锁。所以，每个线程再要获取锁时，可以判定本身的节点序号是否是最小的，假如 是则获取到锁。当释放锁时，只需将本身的暂时有序节点删除即可。
根据上图，在并发下，每个线程都会在对应方法节点下创建属于本身的暂时节点，且每个节点都是暂时且有序的。 那么zookeeper又是怎样有序的将锁分配给不同线程呢？ 这里就应用到了watch监听机制。每当添加一个新的暂时 节点时，其都会基于watcher机制监听着它本身的前一个节点等待前一个节点的通知，当前一个节点删除时，就轮 到它来持有锁了。然后依次类推。
分布式锁的实现
低效锁思想&实现
在通过zookeeper实现分布式锁时，有另外一种实现的写法，这种也黑白常常见的，但是它的效率并不高，此处可以先对这种实现方式举行探究。
此种实现方式，只会存在一个锁节点。当创建锁节点时，假如锁节点不存在，则创建成功，代表当火线程获取到 锁，假如创建锁节点失败，代表已经有其他线程获取到锁，则该线程会监听锁节点的释放。当锁节点释放后，则继 续尝试创建锁节点加锁。
1）在zookeeper_common中创建抽象类AbstractLock
public abstract class AbstractLock {
//zookeeper服务器地址
public static final String ZK_SERVER_ADDR="192.168.200.131:2181";
//zookeeper超时时间
public static final int CONNECTION_TIME_OUT=30000;
public static final int SESSION_TIME_OUT=30000;
//创建zk客户端
protected ZkClient zkClient = new ZkClient(ZK_SERVER_ADDR,SESSION_TIME_OUT,CONNECTION_TIME_OUT);
/**
* 获取锁
* @return
*/
public abstract boolean tryLock();
/**
* 等待加锁
*/
public abstract void waitLock();
/**
* 释放锁
*/
public abstract void releaseLock();
public void getLock() {
String threadName = Thread.currentThread().getName();
if (tryLock()) {
System.out.println(threadName+": 获取锁成功");
}else {
System.out.println(threadName+": 获取锁失败，等待中");
//等待锁
waitLock();
getLock();
}
}
}
复制代码
2）创建LowLock
public class LowLock extends AbstractLock{
private static final String LOCK_NODE_NAME = "/lock_node";
private CountDownLatch countDownLatch;
@Override
public boolean tryLock() {
if (zkClient == null){
return false;
}
try {
zkClient.createEphemeral(LOCK_NODE_NAME);
return true;
} catch (Exception e) {
return false;
}
}
@Override
public void waitLock() {
IZkDataListener zkDataListener = new IZkDataListener() {
//节点被改变时触发
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {
}
//节点被删除时触发
@Override
public void handleDataDeleted(String dataPath) throws Exception {
if (countDownLatch != null){
countDownLatch.countDown();
}
}
};
//注册监听器
zkClient.subscribeDataChanges(LOCK_NODE_NAME,zkDataListener);
//如果锁节点存在，阻塞当前线程
if (zkClient.exists(LOCK_NODE_NAME)){
countDownLatch = new CountDownLatch(1);
try {
countDownLatch.await();
System.out.println(Thread.currentThread().getName()+": 等待获取锁");
} catch (InterruptedException e) {
}
}
//删除监听
zkClient.unsubscribeDataChanges(LOCK_NODE_NAME,zkDataListener);
}
@Override
public void releaseLock() {
zkClient.delete(LOCK_NODE_NAME);
zkClient.close();
System.out.println(Thread.currentThread().getName()+": 释放锁");
}
}
复制代码
3）创建测试类
public class LockTest {
public static void main(String[] args) {
//模拟多个10个客户端
for (int i=0;i<10;i++) {
Thread thread = new Thread(new LockRunnable());
thread.start();
}
}
private static class LockRunnable implements Runnable {
@Override
public void run() {
AbstractLock abstractLock = new LowLock();
abstractLock.getLock();
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
abstractLock.releaseLock();
}
}
}
复制代码
4）颠末测试可以发现，当一个线程获取到锁之后，其他线程都会监听这把锁进入到等待状态，一旦持有锁的线程释放锁后，其他线程则都会监听到，并竞争这把锁。
这种方案的低效点就在于，只有一个锁节点，其他线程都会监听同一个锁节点，一旦锁节点释放后，其他线程都会 收到通知，然后竞争获取锁节点。这种大量的通知操作会严峻降低zookeeper性能，对于这种由于一个被watch的 znode节点的变化，而造成大量的通知操作，叫做羊群效应。
高效锁思想&实现
为了避免羊群效应的出现，业界内广泛的办理方案就是，让获取锁的线程产生排队，后一个监听前一个，依次排 序。保举使用这种方式实现分布式锁
按照上述流程会在根节点下为每一个等待获取锁的线程创建一个对应的暂时有序节点，序号最小的节点会持有锁， 并且后一个节点只监听其前面的一个节点，从而可以让获取锁的过程有序且高效
1）定义HighLock类
public class HighLock extends AbstractLock{
private static final String PARENT_NODE_PATH="/high_lock";
//当前节点路径
private String currentNodePath;
//前一个节点的路径
private String preNodePath;
private CountDownLatch countDownLatch;
@Override
public boolean tryLock() {
//判断父节点是否存在
if (!zkClient.exists(PARENT_NODE_PATH)){
//不存在
zkClient.createPersistent(PARENT_NODE_PATH);
}
//创建第一个临时有序子节点
if (currentNodePath == null || "".equals(currentNodePath)){
//根节点下没有节点信息，将当前节点作为第一个子节点,类型：临时有序
currentNodePath = zkClient.createEphemeralSequential(PARENT_NODE_PATH+"/","lock");
}
//不是第一个子节点，获取父节点下所有子节点
List<String> childrenNodeList = zkClient.getChildren(PARENT_NODE_PATH);
//子节点升序排序
Collections.sort(childrenNodeList);
//判断是否加锁成功
if (currentNodePath.equals(PARENT_NODE_PATH+"/"+childrenNodeList.get(0))){
//当前节点是序号最小的节点
return true;
}else {
//当前节点不是序号最小的节点，获取其前面的节点名称，并赋值
int length = PARENT_NODE_PATH.length();
int currentNodeNumber = Collections.binarySearch(childrenNodeList, currentNodePath.substring(length + 1));
preNodePath = PARENT_NODE_PATH+"/"+childrenNodeList.get(currentNodeNumber-1);
}
return false;
}
@Override
public void waitLock() {
IZkDataListener zkDataListener = new IZkDataListener() {
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {
}
@Override
public void handleDataDeleted(String dataPath) throws Exception {
if (countDownLatch != null){
countDownLatch.countDown();
}
}
};
//监听前一个节点的改变
zkClient.subscribeDataChanges(preNodePath,zkDataListener);
if (zkClient.exists(preNodePath)){
countDownLatch = new CountDownLatch(1);
try {
countDownLatch.await();
} catch (InterruptedException e) {
}
}
zkClient.unsubscribeDataChanges(preNodePath,zkDataListener);
}
@Override
public void releaseLock() {
zkClient.delete(currentNodePath);
zkClient.close();
}
}
复制代码
2）根据结果可以看到，每一个线程都会有本身的节点信息，并且都会有对应的序号。序号最小的节点首先获取到锁，然后依次类推
五、redis分布式锁
详细内容请参考这篇博客
高性能分布式缓存Redis-分布式锁与布隆过滤器_布隆过滤器 分布式锁-CSDN博客
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)
Powered by Discuz! X3.4
