6月安卓手機(jī)性能榜：vivo/iQOO霸占旗艦排行榜前三
2023年上半年已經(jīng)正式過去了，我們也迎來(lái)了安兔兔V10版本，在新的驍龍8Gen3和天璣9300發(fā)布之前，性能榜的榜單大體會(huì)以驍龍8Gen2和天璣9200+為主，至于那顆3.36GHz的驍龍8Gen2領(lǐng)先
在上一篇文章中，我們通過雙異步的方式導(dǎo)入了10萬(wàn)行的Excel，有個(gè)小伙伴在評(píng)論區(qū)問我，如果保證事務(wù)呢，如果分批的話。
通過串行讀取Excel，單個(gè)Excel耗時(shí)191s。
通過Future獲取異步返回值，再和Excel文件數(shù)據(jù)行進(jìn)行比較，實(shí)現(xiàn)對(duì)數(shù)據(jù)準(zhǔn)確性的判斷！
Java8中引入了CompletableFuture，它實(shí)現(xiàn)了對(duì)Future的全面升級(jí)，可以通過回調(diào)的方式，獲取異步線程返回值。
CompletableFuture的異步執(zhí)行通過ForkJoinPool實(shí)現(xiàn)， 它使用守護(hù)線程去執(zhí)行任務(wù)。
ForkJoinPool在于可以充分利用多核CPU的優(yōu)勢(shì)，把一個(gè)任務(wù)拆分成多個(gè)小任務(wù)，把多個(gè)小任務(wù)放到多個(gè)CPU上并行執(zhí)行，當(dāng)多個(gè)小任務(wù)執(zhí)行完畢后，再將其執(zhí)行結(jié)果合并起來(lái)。
想要保證事務(wù)，肯定是使用@Transactional來(lái)實(shí)現(xiàn)。
現(xiàn)在的場(chǎng)景是導(dǎo)入若干個(gè)大的Excel文件數(shù)據(jù)，因?yàn)槊總€(gè)Excel導(dǎo)入的表不同，所以只要保證單Excel的事務(wù)即可。
上文中，是使用異步批量讀取并插入的方式實(shí)現(xiàn)的Excel文件入庫(kù)。
也就是說，1個(gè)主線程事務(wù) + 若干個(gè)子線程事務(wù)，我們想要保證單Excel的插入事務(wù)，所有異步子線程有任何一個(gè)報(bào)錯(cuò)，都要進(jìn)行事務(wù)回滾，如果全部都沒報(bào)錯(cuò)，則進(jìn)行事務(wù)提交。
這個(gè)時(shí)候，有的小伙伴可能會(huì)想到，主線程加個(gè)@Transactional注解，所有子線程分別加@Transactional注解，就可以了吧？
但是，這樣是不行的，子線程的異常只會(huì)回滾其自身的事務(wù)。
如果Excel中有10萬(wàn)條數(shù)據(jù)，一次插入4200條數(shù)據(jù)，最后一次插入3400條。如果其它線程都插入成功了，最后一個(gè)報(bào)錯(cuò)了，此時(shí)，數(shù)據(jù)庫(kù)中還是會(huì)有96600條數(shù)據(jù)插入成功，與單Excel的事務(wù)需求不符。
通過代碼模擬這種情況：
if(end == sheet.getLastRowNum()){ logger.info("插入最后一批數(shù)據(jù)，模擬異常"); int a = 1/0;}
聲明式事務(wù)管理建立在AOP之上的。其本質(zhì)是對(duì)方法前后進(jìn)行攔截，然后在目標(biāo)方法開始之前創(chuàng)建或者加入一個(gè)事務(wù)，在執(zhí)行完目標(biāo)方法之后根據(jù)執(zhí)行情況提交或者回滾事務(wù)。
簡(jiǎn)而言之，@Transactional注解在代碼執(zhí)行出錯(cuò)的時(shí)候能夠進(jìn)行事務(wù)的回滾。
使用@Transactional后，當(dāng)程序發(fā)生RuntimeException運(yùn)行時(shí)異常在沒有使用try，catch進(jìn)行捕獲的時(shí)候，程序都會(huì)中止，當(dāng)程序發(fā)生中止，則會(huì)觸發(fā)數(shù)據(jù)庫(kù)的回滾。
當(dāng)使用了trycatch進(jìn)行捕獲到這個(gè)異常，假如在catch中加入了throw e拋出異常，則程序中止，數(shù)據(jù)庫(kù)回滾。
加入在try catch中沒有throw e 拋出異常，只是簡(jiǎn)單的打印異常，則異常被捕獲未拋出異常去終止程序，在trycatch中的操作數(shù)據(jù)庫(kù)語(yǔ)句插入失敗，在trycatch上面和下面的數(shù)據(jù)庫(kù)相關(guān)插入語(yǔ)句成功，也就是程序成功跑完，數(shù)據(jù)庫(kù)不會(huì)發(fā)生回滾。
在@Transactional注解中如果不配置rollbackFor屬性，那么事物只會(huì)在遇到RuntimeException的時(shí)候才會(huì)回滾，加上rollbackFor=Exception.class，可以讓事物在遇到非運(yùn)行時(shí)異常時(shí)也回滾。
事務(wù)攔截器在目標(biāo)方法執(zhí)行前后進(jìn)行攔截，內(nèi)部會(huì)調(diào)用方法來(lái)獲取Transactional 注解的事務(wù)配置信息，調(diào)用前會(huì)檢查目標(biāo)方法的修飾符是否為 public，不是 public則不會(huì)獲取@Transactional 的屬性配置信息。
rollbackFor 可以指定能夠觸發(fā)事務(wù)回滾的異常類型。
Spring默認(rèn)拋出了未檢查unchecked異常（繼承自 RuntimeException 的異常）或者 Error才回滾事務(wù)；其他異常不會(huì)觸發(fā)回滾事務(wù)。
如果在事務(wù)中拋出其他類型的異常，但卻期望 Spring 能夠回滾事務(wù)，就需要指定rollbackFor屬性。
開發(fā)中避免不了會(huì)對(duì)同一個(gè)類里面的方法調(diào)用，比如有一個(gè)類Test，它的一個(gè)方法A，A再調(diào)用本類的方法B（不論方法B是用public還是private修飾），但方法A沒有聲明注解事務(wù)，而B方法有。則外部調(diào)用方法A之后，方法B的事務(wù)是不會(huì)起作用的。這也是經(jīng)常犯錯(cuò)誤的一個(gè)地方。
那為啥會(huì)出現(xiàn)這種情況？其實(shí)這還是由于使用Spring AOP代理造成的，因?yàn)橹挥挟?dāng)事務(wù)方法被當(dāng)前類以外的代碼調(diào)用時(shí)，才會(huì)由Spring生成的代理對(duì)象來(lái)管理。
在同一個(gè)類中調(diào)用異步方法，等于調(diào)用this本類的方法，沒有走Spring生成的代理類，也就不會(huì)讓他異步執(zhí)行，@Transactional的原理也類似。
如果你手動(dòng)的catch捕獲這個(gè)異常并進(jìn)行處理，事務(wù)管理器會(huì)認(rèn)為當(dāng)前事務(wù)應(yīng)該正常commit，就會(huì)導(dǎo)致注解失效，如果非要捕獲且不失效，就必須在代碼塊內(nèi)throw new Exception拋出異常。
@Transactional(rollbackFor = Exception.class)public void readXls(String filePath, String filename) throws Exception{ try { // 省略一些復(fù)雜操作... List<Future<Integer>> futureList = new ArrayList<>(); for (int time = 0; time < times; time++) { Future<Integer> sumFuture = readExcelDataAsyncFutureService.readXlsCacheAsyncMybatis(); futureList.add(sumFuture); } // 主線程獲取Future返回值 boolean futureFlag = getFutureResult(futureList, excelRow); if (futureFlag) { logger.info("readXlsCacheAsync---插入數(shù)據(jù)成功，提交事務(wù)"); } else { TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(); logger.info("readXlsCacheAsync---插入數(shù)據(jù)失敗，回滾事務(wù)"); } } catch (Exception e) { TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(); logger.error("readXlsCacheAsync---插入數(shù)據(jù)異常，回滾事務(wù)：", e); }}@Async("async-executor")//是否開啟異步@Overridepublic Integer readXlsCacheAsyncMybatis() { try { // 省略一些復(fù)雜操作... }catch (Exception e){ throw new RuntimeException("插入數(shù)據(jù)庫(kù)異常", e); }}如果入庫(kù)異常，事務(wù)回滾成功。
回滾失敗！
public void readXls(String filePath, String filename) throws Exception{ // 手動(dòng)開啟事務(wù),不自動(dòng)提交 TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition); try { // 省略一些復(fù)雜操作... List<Future<Integer>> futureList = new ArrayList<>(); for (int time = 0; time < times; time++) { Future<Integer> sumFuture = readExcelDataAsyncFutureService.readXlsCacheAsyncMybatis(); futureList.add(sumFuture); } // 主線程獲取Future返回值 boolean futureFlag = getFutureResult(futureList, excelRow); if (futureFlag) { dataSourceTransactionManager.commit(transactionStatus); // 提交 logger.info("readXlsCacheAsync---插入數(shù)據(jù)成功，提交事務(wù)"); } else { dataSourceTransactionManager.rollback(transactionStatus);// 回滾 logger.info("readXlsCacheAsync---插入數(shù)據(jù)失敗，回滾事務(wù)"); } } catch (Exception e) { dataSourceTransactionManager.rollback(transactionStatus);// 回滾 logger.error("readXlsCacheAsync---插入數(shù)據(jù)異常，回滾事務(wù)：", e); }}@Async("async-executor")//是否開啟異步@Overridepublic Integer readXlsCacheAsyncMybatis() { try { // 省略一些復(fù)雜操作... }catch (Exception e){ throw new RuntimeException("插入數(shù)據(jù)庫(kù)異常", e); }}如果入庫(kù)異常，事務(wù)回滾成功。
回顧一下需求：異步某線程失敗時(shí)，主線程回滾所有異步線程的事務(wù)！
是代碼有問題，還是就是實(shí)現(xiàn)不了呢？
@Async和@Transactional注解都是通過Spring aop實(shí)現(xiàn)的，核心都是靠著關(guān)鍵的MethodInterceptor實(shí)現(xiàn)，@Async會(huì)給對(duì)應(yīng)bean代理對(duì)象中放入一個(gè)AnnotationAsyncExecutionInterceptor攔截器，而@Transactional會(huì)給對(duì)應(yīng)bean的代理對(duì)象中放入一個(gè)TransactionInterceptor攔截器。
Spring事務(wù)管理的傳播機(jī)制是使用 ThreadLocal 實(shí)現(xiàn)的。因?yàn)?ThreadLocal 是線程私有的，所以 Spring 的事務(wù)傳播機(jī)制是不能夠跨線程的。
/** * 數(shù)據(jù)源事務(wù)管理器 */private DataSourceTransactionManager dataSourceTransactionManager;@Autowiredpublic void setUserService(DataSourceTransactionManager dataSourceTransactionManager) { this.dataSourceTransactionManager = dataSourceTransactionManager;}@Overridepublic void readXls(String filePath, String filename) { List<TransactionStatus> transactionStatusList = Collections.synchronizedList(new ArrayList<>()); List<TransactionResource> transactionResourceList = Collections.synchronizedList(new ArrayList<>()); try { List<Future<Integer>> futureList = new ArrayList<>(); for (int time = 0; time < times; time++) { Future<Integer> sumFuture = readAsyncFutureTransactionDBService.readXlsCacheAsyncMybatis(sheet, row, start, end, insertBuilder,transactionStatusList,transactionResourceList); futureList.add(sumFuture); } // 主線程獲取Future返回值 boolean futureFlag = getFutureResult(futureList, excelRow); if (futureFlag) { for (int i = 0; i < transactionStatusList.size(); i++) { TransactionStatus transactionStatus = transactionStatusList.get(i); dataSourceTransactionManager.commit(transactionStatus); // 提交 } logger.info("readXlsCacheAsync---插入數(shù)據(jù)成功，提交事務(wù)"); } else { for (int i = 0; i < transactionStatusList.size(); i++) { TransactionStatus transactionStatus = transactionStatusList.get(i); dataSourceTransactionManager.rollback(transactionStatus);// 回滾 } logger.info("readXlsCacheAsync---插入數(shù)據(jù)失敗，事務(wù)回滾"); throw new RuntimeException("readXlsCacheAsync---插入數(shù)據(jù)異常，異常事務(wù)回滾"); } } catch (Exception e) { logger.error("readXlsCacheAsync---插入數(shù)據(jù)異常，事務(wù)回滾：", e); for (int i = 0; i < transactionStatusList.size(); i++) { TransactionStatus transactionStatus = transactionStatusList.get(i); dataSourceTransactionManager.rollback(transactionStatus);// 回滾 } //connection.rollback(); throw new RuntimeException("readXlsCacheAsync---插入數(shù)據(jù)異常，異常事務(wù)回滾"); }}@Async("async-executor")@Overridepublic Future<Integer> readXlsCacheAsyncMybatis(XSSFSheet sheet, XSSFRow row, int start, int end, StringBuilder insertBuilder, List<TransactionStatus> transactionStatusList, List<ReadAsyncFutureTransactionServiceImpl.TransactionResource> transactionResourceList) throws Exception { DefaultTransactionDefinition defaultTransactionDefinition = new DefaultTransactionDefinition(); TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(defaultTransactionDefinition); // 開啟新事務(wù) transactionStatusList.add(transactionStatus); // copy事務(wù)資源 transactionResourceList.add(ReadAsyncFutureTransactionServiceImpl.TransactionResource.copyTransactionResource()); try { // 入庫(kù)操作 }catch (Exception e){ throw new RuntimeException("readXlsCacheAsyncMybatis分批異步讀取Excel，通過Mybatis插入數(shù)據(jù)庫(kù)異常"); }}/** * 保存當(dāng)前事務(wù)資源,用于線程間的事務(wù)資源COPY操作 * <p> * `@Builder`注解是Lombok庫(kù)提供的一個(gè)注解,它可以用于自動(dòng)生成Builder模式的代碼,使用@Builder注解可以簡(jiǎn)化創(chuàng)建對(duì)象實(shí)例的過程,并且可以使代碼更加清晰和易于維護(hù) */static class TransactionResource { // TransactionSynchronizationManager類內(nèi)部默認(rèn)提供了下面六個(gè)ThreadLocal屬性,分別保存當(dāng)前線程對(duì)應(yīng)的不同事務(wù)資源 // 保存當(dāng)前事務(wù)關(guān)聯(lián)的資源,默認(rèn)只會(huì)在新建事務(wù)的時(shí)候保存當(dāng)前獲取到的DataSource和當(dāng)前事務(wù)對(duì)應(yīng)Connection的映射關(guān)系 // 當(dāng)然這里Connection被包裝為了ConnectionHolder // 事務(wù)結(jié)束后默認(rèn)會(huì)移除集合中的DataSource作為key關(guān)聯(lián)的資源記錄 private Map<Object, Object> resources; //下面五個(gè)屬性會(huì)在事務(wù)結(jié)束后被自動(dòng)清理,無(wú)需我們手動(dòng)清理 // 事務(wù)監(jiān)聽者,在事務(wù)執(zhí)行到某個(gè)階段的過程中,會(huì)去回調(diào)監(jiān)聽者對(duì)應(yīng)的回調(diào)接口(典型觀察者模式的應(yīng)用),默認(rèn)為空集合 private Set<TransactionSynchronization> synchronizations; // 存放當(dāng)前事務(wù)名字 private String currentTransactionName; // 存放當(dāng)前事務(wù)是否是只讀事務(wù) private Boolean currentTransactionReadOnly; // 存放當(dāng)前事務(wù)的隔離級(jí)別 private Integer currentTransactionIsolationLevel; // 存放當(dāng)前事務(wù)是否處于激活狀態(tài) private Boolean actualTransactionActive; /** * 對(duì)事務(wù)資源進(jìn)行復(fù)制 * * @return TransactionResource */ public static TransactionResource copyTransactionResource() { return TransactionResource.builder() //返回的是不可變集合 .resources(TransactionSynchronizationManager.getResourceMap()) //如果需要注冊(cè)事務(wù)監(jiān)聽者,這里記得修改,我們這里不需要,就采用默認(rèn)負(fù)責(zé),spring事務(wù)內(nèi)部默認(rèn)也是這個(gè)值 .synchronizations(new LinkedHashSet<>()).currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()).currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()).currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()).actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()).build(); } /** * 使用 */ public void autoWiredTransactionResource() { resources.forEach(TransactionSynchronizationManager::bindResource); //如果需要注冊(cè)事務(wù)監(jiān)聽者,這里記得修改,我們這里不需要,就采用默認(rèn)負(fù)責(zé),spring事務(wù)內(nèi)部默認(rèn)也是這個(gè)值 TransactionSynchronizationManager.initSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive); TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName); TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel); TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly); } /** * 移除 */ public void removeTransactionResource() { // 事務(wù)結(jié)束后默認(rèn)會(huì)移除集合中的DataSource作為key關(guān)聯(lián)的資源記錄 // DataSource如果重復(fù)移除,unbindResource時(shí)會(huì)因?yàn)椴淮嬖诖薻ey關(guān)聯(lián)的事務(wù)資源而報(bào)錯(cuò) resources.keySet().forEach(key -> { if (!(key instanceof DataSource)) { TransactionSynchronizationManager.unbindResource(key); } }); }}如何不加會(huì)怎么樣？
在提交和回滾的時(shí)候，會(huì)出現(xiàn)異常：
經(jīng)過不懈的努力，終于解決了“異步某線程失敗時(shí)，主線程回滾所有異步線程的事務(wù)！”這個(gè)看起來(lái)很簡(jiǎn)單的問題。
也是對(duì)雙異步入庫(kù)系列的一個(gè)完結(jié)。
通過添加事務(wù)，可以有效的控制Excel異步插入數(shù)據(jù)的準(zhǔn)確性。
本文鏈接：http://www.www897cc.com/showinfo-26-70399-0.html雙異步系列完結(jié)撒花，如何解決異步事務(wù)問題？
聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請(qǐng)聯(lián)系：2376512515@qq.com 津ICP備18001702號(hào)
津公網(wǎng)安備 12010102000574號(hào)