背景
日常項(xiàng)目中逸雹，經(jīng)常會(huì)出現(xiàn)一個(gè)場(chǎng)景，同時(shí)批量插入數(shù)據(jù)庫數(shù)據(jù)云挟，由于邏輯復(fù)雜或者其它原因梆砸，我們無法使用sql進(jìn)行批量插入。串行效率低园欣，耗時(shí)長辫樱，為了提高效率，這個(gè)時(shí)候我們首先想到多線程并發(fā)插入俊庇，但是如何控制事務(wù)呢 … 直接上干貨
實(shí)現(xiàn)效果
開啟多條子線程狮暑，并發(fā)插入數(shù)據(jù)庫 當(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)榭铡?);
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);
//必須要使用對(duì)象搬男，如果使用變量會(huì)造成線程之間不可共享變量值
RollBack rollBack = new RollBack(false);
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(nThreads);
List<Future<List<Object>>> futures = Lists.newArrayList();
List<Object> returnDataList = Lists.newArrayList();
//給每個(gè)線程分配任務(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);
//如果超時(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 {
//子線程再次開始執(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)用一個(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<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ì)算過程已經(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ù)庫中插入的業(yè)務(wù) 使用 threadLatch.countDown(); 釋放子線程鎖定，同時(shí)使用 mainLatch.await(); 阻塞子線程倾剿，將程序的控制權(quán)交還給主線程 主線程檢查子線程執(zhí)行插入數(shù)據(jù)庫的結(jié)果筷频，若有非預(yù)期結(jié)果出現(xiàn)，主線程標(biāo)記狀態(tài)告知子線程回滾前痘，然后使用 mainLatch.countDown(); 將程序控制權(quán)再次交給子線程凛捏，子線程檢測(cè)回滾標(biāo)志，判斷是否回滾 子線程執(zhí)行結(jié)束芹缔，主線程拼接處理結(jié)果坯癣，響應(yīng)給請(qǐng)求方
整個(gè)過程類似于GC的標(biāo)記-清除過程（串行的垃圾收集器）
