java 对线程进行事务控制_Java 多线程事务回滚 ——多线程插入数据库时事务控制...
背景
日常項(xiàng)目中，經(jīng)常會(huì)出現(xiàn)一個(gè)場(chǎng)景，同時(shí)批量插入數(shù)據(jù)庫(kù)數(shù)據(jù)，由于邏輯復(fù)雜或者其它原因，我們無(wú)法使用sql進(jìn)行批量插入。串行效率低，耗時(shí)長(zhǎng)，為了提高效率，這個(gè)時(shí)候我們首先想到多線程并發(fā)插入，但是如何控制事務(wù)呢 … 直接上干貨
實(shí)現(xiàn)效果
開(kāi)啟多條子線程，并發(fā)插入數(shù)據(jù)庫(kù)
當(dāng)其中一條線程出現(xiàn)異常，或者處理結(jié)果為非預(yù)期結(jié)果，則全部線程均回滾
代碼實(shí)現(xiàn)
@Service
public class CompanyUserBatchServiceImpl implements CompanyUserBatchService {
private static final Logger logger = LoggerFactory.getLogger(CompanyUserBatchServiceImpl.class);
@Autowired
private CompanyUserService companyUserService;
@Override
public ReturnData addNewCurrentCompanyUsers(String params) {
logger.info("addNewCompanyUsers 新增參保人方法");
logger.info(">>>>>>>>>>>>參數(shù):{}", params);
ReturnData rd = new ReturnData();
rd.setRetCode(CommonConstants.RETURN_CODE_FAIL);
if (StringUtils.isBlank(params)) {
rd.setMsg("入?yún)榭?");
logger.info(">>>>>>入?yún)榭铡?#34;);
return rd;
}
List companyUsers;
try {
companyUsers = JSONObject.parseArray(params, CompanyUserResultVo.class);
} catch (Exception e) {
logger.info(">>>>>>>>>入?yún)⒏袷接姓`: {}", e);
rd.setMsg("入?yún)⒏袷接姓`!");
return rd;
}
//每條線程最小處理任務(wù)數(shù)
int perThreadHandleCount = 1;
//線程池的最大線程數(shù)
int nThreads = 10;
int taskSize = companyUsers.size();
if (taskSize > nThreads * perThreadHandleCount) {
perThreadHandleCount = taskSize % nThreads == 0 ? taskSize / nThreads : taskSize / nThreads + 1;
nThreads = taskSize % perThreadHandleCount == 0 ? taskSize / perThreadHandleCount : taskSize / perThreadHandleCount + 1;
} else {
nThreads = taskSize;
}
logger.info("批量添加參保人taskSize: {}, perThreadHandleCount: {}, nThreads: {}", taskSize, perThreadHandleCount, nThreads);
CountDownLatch mainLatch = new CountDownLatch(1);
//監(jiān)控子線程
CountDownLatch threadLatch = new CountDownLatch(nThreads);
//根據(jù)子線程執(zhí)行結(jié)果判斷是否需要回滾
BlockingDeque resultList = new LinkedBlockingDeque<>(nThreads);
//必須要使用對(duì)象，如果使用變量會(huì)造成線程之間不可共享變量值
RollBack rollBack = new RollBack(false);
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(nThreads);
List>> futures = Lists.newArrayList();
List returnDataList = Lists.newArrayList();
//給每個(gè)線程分配任務(wù)
for (int i = 0; i < nThreads; i++) {
int lastIndex = (i + 1) * perThreadHandleCount;
List companyUserResultVos = companyUsers.subList(i * perThreadHandleCount, lastIndex >= taskSize ? taskSize : lastIndex);
AddNewCompanyUserThread addNewCompanyUserThread = new AddNewCompanyUserThread(mainLatch, threadLatch, rollBack, resultList, companyUserResultVos);
Future> future = fixedThreadPool.submit(addNewCompanyUserThread);
futures.add(future);
}
/** 存放子線程返回結(jié)果. */
List backUpResult = Lists.newArrayList();
try {
//等待所有子線程執(zhí)行完畢
boolean await = threadLatch.await(20, TimeUnit.SECONDS);
//如果超時(shí)，直接回滾
if (!await) {
rollBack.setRollBack(true);
} else {
logger.info("創(chuàng)建參保人子線程執(zhí)行完畢，共 {} 個(gè)線程", nThreads);
//查看執(zhí)行情況，如果有存在需要回滾的線程，則全部回滾
for (int i = 0; i < nThreads; i++) {
Boolean result = resultList.take();
backUpResult.add(result);
logger.debug("子線程返回結(jié)果result: {}", result);
if (result) {
/** 有線程執(zhí)行異常，需要回滾子線程. */
rollBack.setRollBack(true);
}
}
}
} catch (InterruptedException e) {
logger.error("等待所有子線程執(zhí)行完畢時(shí)，出現(xiàn)異常");
throw new SystemException("等待所有子線程執(zhí)行完畢時(shí)，出現(xiàn)異常，整體回滾");
} finally {
//子線程再次開(kāi)始執(zhí)行
mainLatch.countDown();
logger.info("關(guān)閉線程池，釋放資源");
fixedThreadPool.shutdown();
}
/** 檢查子線程是否有異常，有異常整體回滾. */
for (int i = 0; i < nThreads; i++) {
if (CollectionUtils.isNotEmpty(backUpResult)) {
Boolean result = backUpResult.get(i);
if (result) {
logger.info("創(chuàng)建參保人失敗，整體回滾");
throw new SystemException("創(chuàng)建參保人失敗");
}
} else {
logger.info("創(chuàng)建參保人失敗，整體回滾");
throw new SystemException("創(chuàng)建參保人失敗");
}
}
//拼接結(jié)果
try {
for (Future> future : futures) {
returnDataList.addAll(future.get());
}
} catch (Exception e) {
logger.info("獲取子線程操作結(jié)果出現(xiàn)異常,創(chuàng)建的參保人列表： {} ，異常信息： {}", JSONObject.toJSONString(companyUsers), e);
throw new SystemException("創(chuàng)建參保人子線程正常創(chuàng)建參保人成功，主線程出現(xiàn)異常，回滾失敗");
}
rd.setRetCode(CommonConstants.RETURN_CODE_SUCCESS);
rd.setData(returnDataList);
return rd;
}
public class AddNewCompanyUserThread implements Callable> {
/**
* 主線程監(jiān)控
*/
private CountDownLatch mainLatch;
/**
* 子線程監(jiān)控
*/
private CountDownLatch threadLatch;
/**
* 是否回滾
*/
private RollBack rollBack;
private BlockingDeque resultList;
private List taskList;
public AddNewCompanyUserThread(CountDownLatch mainLatch, CountDownLatch threadLatch, RollBack rollBack, BlockingDeque resultList, List taskList) {
this.mainLatch = mainLatch;
this.threadLatch = threadLatch;
this.rollBack = rollBack;
this.resultList = resultList;
this.taskList = taskList;
}
@Override
public List call() {
//為了保證事務(wù)不提交，此處只能調(diào)用一個(gè)有事務(wù)的方法，spring 中事務(wù)的顆粒度是方法，只有方法不退出，事務(wù)才不會(huì)提交
return companyUserService.addNewCompanyUsers(mainLatch, threadLatch, rollBack, resultList, taskList);
}
}
public class RollBack {
private Boolean isRollBack;
public Boolean getRollBack() {
return isRollBack;
}
public void setRollBack(Boolean rollBack) {
isRollBack = rollBack;
}
public RollBack(Boolean isRollBack) {
this.isRollBack = isRollBack;
}
}
public List addNewCompanyUsers(CountDownLatch mainLatch, CountDownLatch threadLatch, CompanyUserBatchServiceImpl.RollBack rollBack, BlockingDeque resultList, List taskList) {
List returnDataList = Lists.newArrayList();
Boolean result = false;
logger.debug("線程: {}創(chuàng)建參保人條數(shù) : {}", Thread.currentThread().getName(), taskList.size());
try {
for (CompanyUserResultVo companyUserResultVo : taskList) {
ReturnData returnData = addSingleCompanyUser(companyUserResultVo);
if (returnData.getRetCode() == CommonConstants.RETURN_CODE_FAIL) {
result = true;
}
returnDataList.add(returnData.getData());
}
//Exception 和 Error 都需要抓
} catch (Throwable throwable) {
throwable.printStackTrace();
logger.info("線程: {}創(chuàng)建參保人出現(xiàn)異常： {} ", Thread.currentThread().getName(), throwable);
result = true;
}
resultList.add(result);
threadLatch.countDown();
logger.info("子線程 {} 計(jì)算過(guò)程已經(jīng)結(jié)束，等待主線程通知是否需要回滾", Thread.currentThread().getName());
try {
mainLatch.await();
logger.info("子線程 {} 再次啟動(dòng)", Thread.currentThread().getName());
} catch (InterruptedException e) {
logger.error("批量創(chuàng)建參保人線程InterruptedException異常");
throw new SystemException("批量創(chuàng)建參保人線程InterruptedException異常");
}
if (rollBack.getRollBack()) {
logger.error("批量創(chuàng)建參保人線程回滾, 線程: {}, 需要更新的信息taskList: {}",
Thread.currentThread().getName(),
JSONObject.toJSONString(taskList));
logger.info("子線程 {} 執(zhí)行完畢，線程退出", Thread.currentThread().getName());
throw new SystemException("批量創(chuàng)建參保人線程回滾");
}
logger.info("子線程 {} 執(zhí)行完畢，線程退出", Thread.currentThread().getName());
return returnDataList;
}
思想就是使用兩個(gè)CountDownWatch實(shí)現(xiàn)子線程的二段提交
步驟：
主線程將任務(wù)分發(fā)給子線程，然后 使用 boolean await = threadLatch.await(20,
TimeUnit.SECONDS); 阻塞主線程，等待所有子線程處理向數(shù)據(jù)庫(kù)中插入的業(yè)務(wù)
使用 threadLatch.countDown(); 釋放子線程鎖定，同時(shí)使用 mainLatch.await(); 阻塞子線程，將程序的控制權(quán)交還給主線程
主線程檢查子線程執(zhí)行插入數(shù)據(jù)庫(kù)的結(jié)果，若有非預(yù)期結(jié)果出現(xiàn)，主線程標(biāo)記狀態(tài)告知子線程回滾，然后使用 mainLatch.countDown(); 將程序控制權(quán)再次交給子線程，子線程檢測(cè)回滾標(biāo)志，判斷是否回滾
子線程執(zhí)行結(jié)束，主線程拼接處理結(jié)果，響應(yīng)給請(qǐng)求方
整個(gè)過(guò)程類(lèi)似于GC的標(biāo)記-清除過(guò)程(串行的垃圾收集器)
創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)總結(jié)
以上是生活随笔為你收集整理的java 对线程进行事务控制_Java 多线程事务回滚 ——多线程插入数据库时事务控制...的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： 苹果电脑初使用苹果电脑如何用
- 下一篇： java多线程编程_阿里P8熬到秃头肝出