目录
- 前置
- InsertBatchSuccessServiceImpl.java
- InsertBatchErrorServiceImpl.java
- 效果图
前置
在一些特殊的场景下, 我们需要一些特定的操作.
比如我有一个接口, 做如下操作, 需要保持事物的一致性, 即: 全部成功则提交, 一个异常则全部回滚:
1.insert订单、(耗时1秒)
2.insert订单商品、(耗时1秒)
3.insert子订单、(耗时1秒)
4.insert操作记录、(耗时1秒)
在这波insert操作下来, 就需要花费4秒钟, 那么我们是否可以采用异步的方式进行保存, 将时间保持在1秒钟, 并保持事物一致性
故有了下面的方法
项目地址: https://gitee.com/xmaxm/test-code/tree/master/chaim-mybatis-plus
大致思路
要做异步操作, 就得做多线程, 但是事物是和线程是绑定在一起的,
同时我们知道, commit和rollback是和DML语句一起使用的, 也就是我们能知道这条SQL是成功还是失败
通过上面, 我们就可以进行线程等待, 在所有的DML语句执行之后, 统一进行commit还是rollback, 执行快的等待执行慢的, 当同时OK就进行统一操作
就可采用下面所列的方式, 当然还有很多别的方式也可以进行
强调:
多测, 做线程循环测试跑. 比如AB, JMeter, apifox, 下面列举的InsertBatchErrorServiceImpl在普通测试过程中不会出现问题, 但是当次数过多就会出现无法唤醒的情况. 一定要测试, 有些场景需要多跑几遍才能够进行重现
代码部分
InsertBatchSuccessServiceImpl:
该实现采用的是CountDownLatch, countDown()递减锁的数量, await()等待直到当前计数器数量为0, 释放所有等待线程
InsertBatchErrorServiceImpl:
该实现采用的 LockSupport.park()悬停、LockSupport.unpark(thread)唤醒. 但在实际使用过程中发现会出现无法唤醒的情况, 我发布了问题(可供参考 ), 但是目前还没有得到解决, 不得不暂时放弃
InsertBatchSuccessServiceImpl.java
package com.chaim.mybatis.service.impl;
import com.chaim.mybatis.converter.SysUserConverter;
import com.chaim.mybatis.dto.SysUserDTO;
import com.chaim.mybatis.entitys.SysUser;
import com.chaim.mybatis.mappers.SysUserMapper;
import com.chaim.mybatis.service.InsertBatchService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Chaim
* @date 2022/12/24 16:06
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class InsertBatchSuccessServiceImpl implements InsertBatchService {
private final SysUserMapper sysUserMapper;
private final SysUserConverter sysUserConverter;
private final DataSourceTransactionManager dataSourceTransactionManager;
@Override
public Object insertBatch(SysUserDTO.InsertSysUserDTO insertSysUserDTO) {
// 定义开启的线程数
final int i = 3;
CountDownLatch latch = new CountDownLatch(i);
// 事务定义
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
// 所有开启事务的线程中, 是否存在异常
AtomicBoolean isException = new AtomicBoolean(Boolean.FALSE);
List<CompletableFuture<Void>> list = new ArrayList<>();
list.add(this.save(insertSysUserDTO, latch, definition, isException));
list.add(this.save1(insertSysUserDTO, latch, definition, isException));
list.add(this.save2(insertSysUserDTO, latch, definition, isException));
// 调用Future的阻塞接口, 等待全部future实例异步执行结束
CompletableFuture.allOf(list.toArray(new CompletableFuture[i])).join();
return "SUCCESS";
}
private CompletableFuture<Void> save(SysUserDTO.InsertSysUserDTO insertSysUserDTO, CountDownLatch latch, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
// 获得事务状态
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(1234);
sysUserMapper.insert(sysUser);
this.threadBlocking(latch, status, isException);
} catch (Exception exception) {
log.error("方法: [save] 异常: {}", exception.getMessage());
this.errorRollback(latch, status, isException);
}
});
}
private CompletableFuture<Void> save1(SysUserDTO.InsertSysUserDTO insertSysUserDTO, CountDownLatch latch, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(9876);
sysUser.setUsername(sysUser.getUsername().concat(": 子线程1"));
sysUserMapper.insert(sysUser);
this.threadBlocking(latch, status, isException);
} catch (Exception exception) {
log.error("方法: [save1] 异常: {}", exception.getMessage());
this.errorRollback(latch, status, isException);
}
});
}
private CompletableFuture<Void> save2(SysUserDTO.InsertSysUserDTO insertSysUserDTO, CountDownLatch latch, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(9876);
sysUser.setUsername(sysUser.getUsername().concat(": 子线程2"));
sysUserMapper.insert(sysUser);
this.threadBlocking(latch, status, isException);
} catch (Exception exception) {
log.error("方法: [save2] 异常: {}", exception.getMessage());
this.errorRollback(latch, status, isException);
}
});
}
/**
* 进行线程阻塞操作
*
* @param latch
* @param status
* @param isException
*/
private void threadBlocking(CountDownLatch latch, TransactionStatus status, AtomicBoolean isException) throws InterruptedException {
log.info("计数器递减");
latch.countDown();
log.info("开始悬停, 剩余计数数量: {}", latch.getCount());
latch.await();
// 以下步骤抛出异常进入catch, 做countDown操作, 不会影响. 走到这一步已经没有阻塞了
if (isException.get()) {
log.info("开始回滚");
dataSourceTransactionManager.rollback(status);
} else {
log.info("开始提交");
dataSourceTransactionManager.commit(status);
}
}
/**
* 程序异常, 进行回滚, 线程唤醒
*
* @param latch
* @param status
* @param isException
*/
private void errorRollback(CountDownLatch latch, TransactionStatus status, AtomicBoolean isException) {
// 设定线程中存在异常信息
isException.set(Boolean.TRUE);
latch.countDown();
log.info("开始回滚, 程序异常, 计数器递减, 剩余数量: {}", latch.getCount());
// 本线程回滚
dataSourceTransactionManager.rollback(status);
}
}
InsertBatchErrorServiceImpl.java
关于该方式存在的问题, 我已经在问答区提出疑问, 可供参考
package com.chaim.mybatis.service.impl;
import com.chaim.mybatis.converter.SysUserConverter;
import com.chaim.mybatis.dto.SysUserDTO;
import com.chaim.mybatis.entitys.SysUser;
import com.chaim.mybatis.mappers.SysUserMapper;
import com.chaim.mybatis.service.InsertBatchService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
/**
* @author Chaim
* @date 2022/12/24 22:06
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class InsertBatchErrorServiceImpl implements InsertBatchService {
private final SysUserMapper sysUserMapper;
private final SysUserConverter sysUserConverter;
private final DataSourceTransactionManager dataSourceTransactionManager;
@Override
public Object insertBatch(SysUserDTO.InsertSysUserDTO insertSysUserDTO) {
// 定义开启的线程数
AtomicInteger totalThreadCount = new AtomicInteger(3);
// 事务定义
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
// 所有开启事务的线程中, 是否存在异常
AtomicBoolean isException = new AtomicBoolean(Boolean.FALSE);
// 定义没有执行结束的线程集合
List<Thread> unFinishedThread = Collections.synchronizedList(new ArrayList<>());
List<CompletableFuture<Void>> list = new ArrayList<>();
list.add(this.save(insertSysUserDTO, unFinishedThread, totalThreadCount, definition, isException));
list.add(this.save1(insertSysUserDTO, unFinishedThread, totalThreadCount, definition, isException));
list.add(this.save2(insertSysUserDTO, unFinishedThread, totalThreadCount, definition, isException));
// 调用Future的阻塞接口, 等待全部future实例异步执行结束
CompletableFuture.allOf(list.toArray(new CompletableFuture[totalThreadCount.get()])).join();
return "SUCCESS";
}
private CompletableFuture<Void> save(SysUserDTO.InsertSysUserDTO insertSysUserDTO, List<Thread> unFinishedThread, AtomicInteger totalThreadCount, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
// 获得事务状态
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(1234);
sysUserMapper.insert(sysUser);
this.threadBlocking(Thread.currentThread(), unFinishedThread, totalThreadCount, status, isException);
} catch (Exception exception) {
log.error("方法: [save] 异常: {}", exception.getMessage());
this.errorRollback(unFinishedThread, totalThreadCount, status, isException);
}
});
}
private CompletableFuture<Void> save1(SysUserDTO.InsertSysUserDTO insertSysUserDTO, List<Thread> unFinishedThread, AtomicInteger totalThreadCount, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(9876);
sysUser.setUsername(sysUser.getUsername().concat(": 子线程1"));
sysUserMapper.insert(sysUser);
this.threadBlocking(Thread.currentThread(), unFinishedThread, totalThreadCount, status, isException);
} catch (Exception exception) {
log.error("方法: [save1] 异常: {}", exception.getMessage());
this.errorRollback(unFinishedThread, totalThreadCount, status, isException);
}
});
}
private CompletableFuture<Void> save2(SysUserDTO.InsertSysUserDTO insertSysUserDTO, List<Thread> unFinishedThread, AtomicInteger totalThreadCount, DefaultTransactionDefinition definition, AtomicBoolean isException) {
return CompletableFuture.runAsync(() -> {
TransactionStatus status = dataSourceTransactionManager.getTransaction(definition);
SysUser sysUser = sysUserConverter.insertUserDTOToSysUser(insertSysUserDTO);
try {
sysUser.setPassword("123456");
sysUser.setSalt(9876);
sysUser.setUsername(sysUser.getUsername().concat(": 子线程2"));
sysUserMapper.insert(sysUser);
this.threadBlocking(Thread.currentThread(), unFinishedThread, totalThreadCount, status, isException);
} catch (Exception exception) {
log.error("方法: [save2] 异常: {}", exception.getMessage());
this.errorRollback(unFinishedThread, totalThreadCount, status, isException);
}
});
}
/**
* 进行线程阻塞操作
*
* @param thread
* @param unFinishedThread
* @param totalThreadCount
* @param status
* @param isException
*/
private void threadBlocking(Thread thread, List<Thread> unFinishedThread, AtomicInteger totalThreadCount, TransactionStatus status, AtomicBoolean isException) {
// 添加到没有执行结束的线程集合
unFinishedThread.add(thread);
// 每个线程都在悬停前开启唤醒检查
this.notifyAllThread(unFinishedThread, totalThreadCount, false);
if (isException.get()) {
log.info("已存在异步任务发生回滚, 当前线程: {}", thread.getName());
dataSourceTransactionManager.rollback(status);
} else {
log.info("线程: {}, 开始悬停", thread.getName());
LockSupport.park();
if (isException.get()) {
log.info("线程: {}, 开始回滚", thread.getName());
dataSourceTransactionManager.rollback(status);
} else {
log.info("线程: {}, 开始提交", thread.getName());
dataSourceTransactionManager.commit(status);
}
}
}
/**
* 程序异常, 进行回滚, 线程唤醒
*
* @param unFinishedThread
* @param totalThreadCount
* @param status
* @param isException
*/
private void errorRollback(List<Thread> unFinishedThread, AtomicInteger totalThreadCount, TransactionStatus status, AtomicBoolean isException) {
// 设定线程中存在异常信息
isException.set(Boolean.TRUE);
// 本线程回滚
dataSourceTransactionManager.rollback(status);
// 发生异常, 全部线程进行唤醒
this.notifyAllThread(unFinishedThread, totalThreadCount, true);
log.info("异常回滚, 开始全部线程唤醒, 当前线程数量: {}", unFinishedThread.size());
}
/**
* 唤醒全部悬停的线程
*
* @param unFinishedThread 手动悬停的线程
* @param totalThreadCount 全部开启的线程数
* @param isForce 是否强行操作集合中全部线程
*/
private void notifyAllThread(List<Thread> unFinishedThread, AtomicInteger totalThreadCount, boolean isForce) {
if (isForce || unFinishedThread.size() == totalThreadCount.get()) {
for (Thread thread : unFinishedThread) {
LockSupport.unpark(thread);
log.info("线程: [{}]被唤醒", thread.getName());
}
}
}
}