日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當前位置:首頁 > 科技  > 軟件

雙異步系列完結撒花,如何解決異步事務問題?

來源: 責編: 時間:2024-02-01 12:46:24 200觀看
導讀一、前情提要在上一篇文章中,我們通過雙異步的方式導入了10萬行的Excel,有個小伙伴在評論區問我,如果保證事務呢,如果分批的話。原始需求:讀取一個10萬行的Excel通過串行讀取Excel,單個Excel耗時191s。優化1:使用雙異步后,從

0pA28資訊網——每日最新資訊28at.com

一、前情提要

在上一篇文章中,我們通過雙異步的方式導入了10萬行的Excel,有個小伙伴在評論區問我,如果保證事務呢,如果分批的話。0pA28資訊網——每日最新資訊28at.com

原始需求:讀取一個10萬行的Excel

通過串行讀取Excel,單個Excel耗時191s。0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

優化1:使用雙異步后,從 191s 優化到 2s

  • 分別通過POI和EasyExcel的方式讀取Excel并插入數據庫。
  • 探討了“線程池中的核心線程數設置問題”。
  • 經過數十次的測試,總結了通過線程池的方式,爭取一次性并行入庫,效率最佳。

0pA28資訊網——每日最新資訊28at.com

優化2:使用雙異步后,如何保證數據一致性?

通過Future獲取異步返回值,再和Excel文件數據行進行比較,實現對數據準確性的判斷!0pA28資訊網——每日最新資訊28at.com

  • 逐行分析了FutureTask源碼,繪制了FutureTask執行流程圖。
  • 分析get()源碼,繪制get()方法執行流程圖。
  • 但是,發現了一個問題,Future.get()會造成主線程的阻塞。

0pA28資訊網——每日最新資訊28at.com

優化3:獲取雙異步返回值時,如何保證主線程不阻塞?

Java8中引入了CompletableFuture,它實現了對Future的全面升級,可以通過回調的方式,獲取異步線程返回值。0pA28資訊網——每日最新資訊28at.com

CompletableFuture的異步執行通過ForkJoinPool實現, 它使用守護線程去執行任務。0pA28資訊網——每日最新資訊28at.com

ForkJoinPool在于可以充分利用多核CPU的優勢,把一個任務拆分成多個小任務,把多個小任務放到多個CPU上并行執行,當多個小任務執行完畢后,再將其執行結果合并起來。0pA28資訊網——每日最新資訊28at.com

  • 通過CompletableFuture優化 “通過Future獲取異步返回值”;
  • CompletableFuture和Future的效率對比;
  • 自定義ForkJoinPool線程池;
  • 核心線程數相同的情況下,CompletableFuture的入庫效率要優于Future的入庫效率,10萬條數據大概要快4秒鐘;
  • 通過CompletableFuture.allOf解決阻塞主線程問題;
  • 總結了CompletableFuture中花俏的語法糖。

二、異步某線程失敗時,主線程回滾所有異步線程的事務!

想要保證事務,肯定是使用@Transactional來實現。0pA28資訊網——每日最新資訊28at.com

現在的場景是導入若干個大的Excel文件數據,因為每個Excel導入的表不同,所以只要保證單Excel的事務即可。0pA28資訊網——每日最新資訊28at.com

上文中,是使用異步批量讀取并插入的方式實現的Excel文件入庫。0pA28資訊網——每日最新資訊28at.com

也就是說,1個主線程事務 + 若干個子線程事務,我們想要保證單Excel的插入事務,所有異步子線程有任何一個報錯,都要進行事務回滾,如果全部都沒報錯,則進行事務提交。0pA28資訊網——每日最新資訊28at.com

這個時候,有的小伙伴可能會想到,主線程加個@Transactional注解,所有子線程分別加@Transactional注解,就可以了吧?0pA28資訊網——每日最新資訊28at.com

但是,這樣是不行的,子線程的異常只會回滾其自身的事務。0pA28資訊網——每日最新資訊28at.com

如果Excel中有10萬條數據,一次插入4200條數據,最后一次插入3400條。如果其它線程都插入成功了,最后一個報錯了,此時,數據庫中還是會有96600條數據插入成功,與單Excel的事務需求不符。0pA28資訊網——每日最新資訊28at.com

通過代碼模擬這種情況:0pA28資訊網——每日最新資訊28at.com

if(end == sheet.getLastRowNum()){    logger.info("插入最后一批數據,模擬異常");    int a = 1/0;}

0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

三、@Transactional注解

聲明式事務管理建立在AOP之上的。其本質是對方法前后進行攔截,然后在目標方法開始之前創建或者加入一個事務,在執行完目標方法之后根據執行情況提交或者回滾事務。0pA28資訊網——每日最新資訊28at.com

簡而言之,@Transactional注解在代碼執行出錯的時候能夠進行事務的回滾。0pA28資訊網——每日最新資訊28at.com

  • 在啟動類上添加@EnableTransactionManagement注解。
  • 用于類上時,該類的所有 public 方法將都具有該類型的事務屬性,同時,我們也可以在方法級別使用該標注來覆蓋類級別的定義。
  • 在項目中,@Transactional(rollbackFor=Exception.class),如果類加了這個注解,那么這個類里面的方法拋出異常,就會回滾,數據庫里面的數據也會回滾。
  • 在@Transactional注解中如果不配置rollbackFor屬性,那么事物只會在遇到RuntimeException的時候才會回滾,加上rollbackFor=Exception.class,可以讓事物在遇到非運行時異常時也回滾。

1、@Transactional

使用@Transactional后,當程序發生RuntimeException運行時異常在沒有使用try,catch進行捕獲的時候,程序都會中止,當程序發生中止,則會觸發數據庫的回滾。0pA28資訊網——每日最新資訊28at.com

當使用了trycatch進行捕獲到這個異常,假如在catch中加入了throw e拋出異常,則程序中止,數據庫回滾。0pA28資訊網——每日最新資訊28at.com

加入在try catch中沒有throw e 拋出異常,只是簡單的打印異常,則異常被捕獲未拋出異常去終止程序,在trycatch中的操作數據庫語句插入失敗,在trycatch上面和下面的數據庫相關插入語句成功,也就是程序成功跑完,數據庫不會發生回滾。0pA28資訊網——每日最新資訊28at.com

2、@Transactional(rollbackFor = Exception.class)

在@Transactional注解中如果不配置rollbackFor屬性,那么事物只會在遇到RuntimeException的時候才會回滾,加上rollbackFor=Exception.class,可以讓事物在遇到非運行時異常時也回滾。0pA28資訊網——每日最新資訊28at.com

四、注解失效問題

1、@Transactional 應用在非 public 修飾的方法上

事務攔截器在目標方法執行前后進行攔截,內部會調用方法來獲取Transactional 注解的事務配置信息,調用前會檢查目標方法的修飾符是否為 public,不是 public則不會獲取@Transactional 的屬性配置信息。0pA28資訊網——每日最新資訊28at.com

2、@Transactional 注解屬性 rollbackFor 設置錯誤

rollbackFor 可以指定能夠觸發事務回滾的異常類型。0pA28資訊網——每日最新資訊28at.com

Spring默認拋出了未檢查unchecked異常(繼承自 RuntimeException 的異常)或者 Error才回滾事務;其他異常不會觸發回滾事務。0pA28資訊網——每日最新資訊28at.com

如果在事務中拋出其他類型的異常,但卻期望 Spring 能夠回滾事務,就需要指定rollbackFor屬性。0pA28資訊網——每日最新資訊28at.com

3、同一個類中方法調用,導致@Transactional失效

開發中避免不了會對同一個類里面的方法調用,比如有一個類Test,它的一個方法A,A再調用本類的方法B(不論方法B是用public還是private修飾),但方法A沒有聲明注解事務,而B方法有。則外部調用方法A之后,方法B的事務是不會起作用的。這也是經常犯錯誤的一個地方。0pA28資訊網——每日最新資訊28at.com

那為啥會出現這種情況?其實這還是由于使用Spring AOP代理造成的,因為只有當事務方法被當前類以外的代碼調用時,才會由Spring生成的代理對象來管理。0pA28資訊網——每日最新資訊28at.com

在同一個類中調用異步方法,等于調用this本類的方法,沒有走Spring生成的代理類,也就不會讓他異步執行,@Transactional的原理也類似。0pA28資訊網——每日最新資訊28at.com

4、捕獲異常

如果你手動的catch捕獲這個異常并進行處理,事務管理器會認為當前事務應該正常commit,就會導致注解失效,如果非要捕獲且不失效,就必須在代碼塊內throw new Exception拋出異常。0pA28資訊網——每日最新資訊28at.com

五、通過Future獲取異步返回值,添加事務

1、添加事務

@Transactional(rollbackFor = Exception.class)public void readXls(String filePath, String filename) throws Exception{ try {  // 省略一些復雜操作...      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---插入數據成功,提交事務");        } else {            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();            logger.info("readXlsCacheAsync---插入數據失敗,回滾事務");        } } catch (Exception e) {  TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  logger.error("readXlsCacheAsync---插入數據異常,回滾事務:", e);    }}@Async("async-executor")//是否開啟異步@Overridepublic Integer readXlsCacheAsyncMybatis() {    try {     // 省略一些復雜操作...    }catch (Exception e){        throw new RuntimeException("插入數據庫異常", e);    }}

(1)添加事務 + 不開啟異步

如果入庫異常,事務回滾成功。0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

(2)添加事務 + 開啟異步

回滾失敗!0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

2、手動添加事務

public void readXls(String filePath, String filename) throws Exception{  // 手動開啟事務,不自動提交    TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition); try {  // 省略一些復雜操作...      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---插入數據成功,提交事務");        } else {            dataSourceTransactionManager.rollback(transactionStatus);// 回滾            logger.info("readXlsCacheAsync---插入數據失敗,回滾事務");        } } catch (Exception e) {  dataSourceTransactionManager.rollback(transactionStatus);// 回滾  logger.error("readXlsCacheAsync---插入數據異常,回滾事務:", e);    }}@Async("async-executor")//是否開啟異步@Overridepublic Integer readXlsCacheAsyncMybatis() {    try {     // 省略一些復雜操作...    }catch (Exception e){        throw new RuntimeException("插入數據庫異常", e);    }}

(1)添加事務 + 不開啟異步

如果入庫異常,事務回滾成功。0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

(2)Future獲取異步返回值,添加手動事務,異常回滾失敗!

0pA28資訊網——每日最新資訊28at.com

六、@async + @Transactional 事務失效問題

回顧一下需求:異步某線程失敗時,主線程回滾所有異步線程的事務!0pA28資訊網——每日最新資訊28at.com

是代碼有問題,還是就是實現不了呢?0pA28資訊網——每日最新資訊28at.com

@Async和@Transactional注解都是通過Spring aop實現的,核心都是靠著關鍵的MethodInterceptor實現,@Async會給對應bean代理對象中放入一個AnnotationAsyncExecutionInterceptor攔截器,而@Transactional會給對應bean的代理對象中放入一個TransactionInterceptor攔截器。0pA28資訊網——每日最新資訊28at.com

Spring事務管理的傳播機制是使用 ThreadLocal 實現的。因為 ThreadLocal 是線程私有的,所以 Spring 的事務傳播機制是不能夠跨線程的。0pA28資訊網——每日最新資訊28at.com

七、Spring 的事務傳播機制是不能夠跨線程的

1、一個異步線程一個事務,然后根據結果統一提交/回滾?

2、核心代碼

/** * 數據源事務管理器 */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---插入數據成功,提交事務");  } else {   for (int i = 0; i < transactionStatusList.size(); i++) {    TransactionStatus transactionStatus = transactionStatusList.get(i);    dataSourceTransactionManager.rollback(transactionStatus);// 回滾   }   logger.info("readXlsCacheAsync---插入數據失敗,事務回滾");   throw new RuntimeException("readXlsCacheAsync---插入數據異常,異常事務回滾");  } } catch (Exception e) {  logger.error("readXlsCacheAsync---插入數據異常,事務回滾:", e);  for (int i = 0; i < transactionStatusList.size(); i++) {   TransactionStatus transactionStatus = transactionStatusList.get(i);   dataSourceTransactionManager.rollback(transactionStatus);// 回滾  }  //connection.rollback();  throw new RuntimeException("readXlsCacheAsync---插入數據異常,異常事務回滾"); }}

3、異步線程類

@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); // 開啟新事務 transactionStatusList.add(transactionStatus); // copy事務資源 transactionResourceList.add(ReadAsyncFutureTransactionServiceImpl.TransactionResource.copyTransactionResource()); try {  // 入庫操作 }catch (Exception e){  throw new RuntimeException("readXlsCacheAsyncMybatis分批異步讀取Excel,通過Mybatis插入數據庫異常"); }}

4、事務復制類

/** * 保存當前事務資源,用于線程間的事務資源COPY操作 * <p> * `@Builder`注解是Lombok庫提供的一個注解,它可以用于自動生成Builder模式的代碼,使用@Builder注解可以簡化創建對象實例的過程,并且可以使代碼更加清晰和易于維護 */static class TransactionResource {    // TransactionSynchronizationManager類內部默認提供了下面六個ThreadLocal屬性,分別保存當前線程對應的不同事務資源    // 保存當前事務關聯的資源,默認只會在新建事務的時候保存當前獲取到的DataSource和當前事務對應Connection的映射關系    // 當然這里Connection被包裝為了ConnectionHolder    // 事務結束后默認會移除集合中的DataSource作為key關聯的資源記錄    private Map<Object, Object> resources;    //下面五個屬性會在事務結束后被自動清理,無需我們手動清理    // 事務監聽者,在事務執行到某個階段的過程中,會去回調監聽者對應的回調接口(典型觀察者模式的應用),默認為空集合    private Set<TransactionSynchronization> synchronizations;    // 存放當前事務名字    private String currentTransactionName;    // 存放當前事務是否是只讀事務    private Boolean currentTransactionReadOnly;    // 存放當前事務的隔離級別    private Integer currentTransactionIsolationLevel;    // 存放當前事務是否處于激活狀態    private Boolean actualTransactionActive;    /**     * 對事務資源進行復制     *     * @return TransactionResource     */    public static TransactionResource copyTransactionResource() {        return TransactionResource.builder()                //返回的是不可變集合                .resources(TransactionSynchronizationManager.getResourceMap())                //如果需要注冊事務監聽者,這里記得修改,我們這里不需要,就采用默認負責,spring事務內部默認也是這個值                .synchronizations(new LinkedHashSet<>()).currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()).currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()).currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()).actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()).build();    }    /**     * 使用     */    public void autoWiredTransactionResource() {        resources.forEach(TransactionSynchronizationManager::bindResource);        //如果需要注冊事務監聽者,這里記得修改,我們這里不需要,就采用默認負責,spring事務內部默認也是這個值        TransactionSynchronizationManager.initSynchronization();        TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);        TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);        TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);        TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);    }    /**     * 移除     */    public void removeTransactionResource() {        // 事務結束后默認會移除集合中的DataSource作為key關聯的資源記錄        // DataSource如果重復移除,unbindResource時會因為不存在此key關聯的事務資源而報錯        resources.keySet().forEach(key -> {            if (!(key instanceof DataSource)) {                TransactionSynchronizationManager.unbindResource(key);            }        });    }}

5、為何要用事務復制類?而最后提交和回滾的時候也沒用它?

如何不加會怎么樣?0pA28資訊網——每日最新資訊28at.com

在提交和回滾的時候,會出現異常:0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

八、總結

經過不懈的努力,終于解決了“異步某線程失敗時,主線程回滾所有異步線程的事務!”這個看起來很簡單的問題。0pA28資訊網——每日最新資訊28at.com

也是對雙異步入庫系列的一個完結。0pA28資訊網——每日最新資訊28at.com

通過添加事務,可以有效的控制Excel異步插入數據的準確性。0pA28資訊網——每日最新資訊28at.com

0pA28資訊網——每日最新資訊28at.com

讀取一個10萬行的Excel的最佳解決方案是:

  • 通過EasyExcel異步讀取Excel;
  • 通過Future獲取異步返回值,比較Excel行數和入庫數,保證數據入庫一致性;
  • 通過CompletableFuture + 自定義ForkJoinPool線程池的方式執行,解決主線程阻塞問題;
  • 根據核心線程數,設置每個線程讀取的Excel數據行數,以達到效率最佳;
  • 通過手動添加事務 + 一個線程一個事務 + 復制事務的方式實現異步事務的有效控制。

0pA28資訊網——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-70399-0.html雙異步系列完結撒花,如何解決異步事務問題?

聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com

上一篇: 如何在 Npm 上發布二進制文件?

下一篇: 項目明明部署成功了, BUG 怎么還在啊?產品急了

標簽:
  • 熱門焦點
Top 主站蜘蛛池模板: 顺昌县| 蓬安县| 通河县| 丰台区| 晋中市| 崇文区| 凤庆县| 芒康县| 齐河县| 志丹县| 双峰县| 英山县| 葵青区| 高州市| 达日县| 孝义市| 龙井市| 灵山县| 信宜市| 泸定县| 都江堰市| 余干县| 阜阳市| 南漳县| 南华县| 雅江县| 京山县| 新巴尔虎左旗| 湘潭市| 新平| 普安县| 江阴市| 喀喇| 长葛市| 紫云| 恩施市| 忻城县| 和顺县| 新蔡县| 盐池县| 星子县|