背景
日常項目中，經(jīng)常會出現(xiàn)一個場景鉴嗤，同時批量插入數(shù)據(jù)庫數(shù)據(jù)斩启，由于邏輯復(fù)雜或者其它原因，我們無法使用sql進行批量插入躬窜。串行效率低浇垦，耗時長炕置，為了提高效率荣挨，這個時候我們首先想到多線程并發(fā)插入，但是如何控制事務(wù)呢 … 直接上干貨
實現(xiàn)效果
開啟多條子線程朴摊，并發(fā)插入數(shù)據(jù)庫
當(dāng)其中一條線程出現(xiàn)異常默垄，或者處理結(jié)果為非預(yù)期結(jié)果，則全部線程均回滾
代碼實現(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)榭铡?);
return rd;
}
List<CompanyUserResultVo> 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<Boolean> resultList = new LinkedBlockingDeque<>(nThreads);
//必須要使用對象甚纲，如果使用變量會造成線程之間不可共享變量值
RollBack rollBack = new RollBack(false);
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(nThreads);
List<Future<List<Object>>> futures = Lists.newArrayList();
List<Object> returnDataList = Lists.newArrayList();
//給每個線程分配任務(wù)
for (int i = 0; i < nThreads; i++) {
int lastIndex = (i + 1) * perThreadHandleCount;
List<CompanyUserResultVo> companyUserResultVos = companyUsers.subList(i * perThreadHandleCount, lastIndex >= taskSize ? taskSize : lastIndex);
AddNewCompanyUserThread addNewCompanyUserThread = new AddNewCompanyUserThread(mainLatch, threadLatch, rollBack, resultList, companyUserResultVos);
Future<List<Object>> future = fixedThreadPool.submit(addNewCompanyUserThread);
futures.add(future);
}
/** 存放子線程返回結(jié)果. */
List<Boolean> backUpResult = Lists.newArrayList();
try {
//等待所有子線程執(zhí)行完畢
boolean await = threadLatch.await(20, TimeUnit.SECONDS);
//如果超時口锭，直接回滾
if (!await) {
rollBack.setRollBack(true);
} else {
logger.info("創(chuàng)建參保人子線程執(zhí)行完畢，共 {} 個線程", 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í)行完畢時春哨，出現(xiàn)異常");
throw new SystemException("等待所有子線程執(zhí)行完畢時荆隘，出現(xiàn)異常，整體回滾");
} finally {
//子線程再次開始執(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<List<Object>> 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<List<Object>> {
/**
* 主線程監(jiān)控
*/
private CountDownLatch mainLatch;
/**
* 子線程監(jiān)控
*/
private CountDownLatch threadLatch;
/**
* 是否回滾
*/
private RollBack rollBack;
private BlockingDeque<Boolean> resultList;
private List<CompanyUserResultVo> taskList;
public AddNewCompanyUserThread(CountDownLatch mainLatch, CountDownLatch threadLatch, RollBack rollBack, BlockingDeque<Boolean> resultList, List<CompanyUserResultVo> taskList) {
this.mainLatch = mainLatch;
this.threadLatch = threadLatch;
this.rollBack = rollBack;
this.resultList = resultList;
this.taskList = taskList;
}
@Override
public List<Object> call() {
//為了保證事務(wù)不提交缆毁，此處只能調(diào)用一個有事務(wù)的方法番川，spring 中事務(wù)的顆粒度是方法，只有方法不退出积锅，事務(wù)才不會提交
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<Object> addNewCompanyUsers(CountDownLatch mainLatch, CountDownLatch threadLatch, CompanyUserBatchServiceImpl.RollBack rollBack, BlockingDeque<Boolean> resultList, List<CompanyUserResultVo> taskList) {
List<Object> 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īng)結(jié)束爽彤，等待主線程通知是否需要回滾", Thread.currentThread().getName());
try {
mainLatch.await();
logger.info("子線程 {} 再次啟動", 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;
}
思想就是使用兩個CountDownWatch實現(xiàn)子線程的二段提交
步驟：
1适篙、主線程將任務(wù)分發(fā)給子線程，然后使用 boolean await = threadLatch.await(20,TimeUnit.SECONDS);阻塞主線程箫爷，等待所有子線程處理向數(shù)據(jù)庫中插入的業(yè)務(wù)
2嚷节、使用threadLatch.countDown();釋放子線程鎖定，同時使用mainLatch.await();阻塞子線程虎锚，將程序的控制權(quán)交還給主線程硫痰。
3、主線程檢查子線程執(zhí)行插入數(shù)據(jù)庫的結(jié)果窜护，若有非預(yù)期結(jié)果出現(xiàn)效斑，主線程標(biāo)記狀態(tài)告知子線程回滾，然后使用mainLatch.countDown();將程序控制權(quán)再次交給子線程柱徙，子線程檢測回滾標(biāo)志缓屠，判斷是否回滾。
4护侮、子線程執(zhí)行結(jié)束敌完，主線程拼接處理結(jié)果，響應(yīng)給請求方
