Spring @Async之三：Spring @Async使用方法總結

引言： 在Java應用中，絕大多數情況下都是通過同步的方式來實現交互處理的；但是在處理與第三方系統交互的時候，容易造成響應遲緩的情況，之前大部分都是使用多線程來完成此類任務，其實，在spring 3.x之后，就已經內置了@Async來完美解決這個問題，本文將完成介紹@Async的用法。

1.  何為異步調用？

    在解釋異步調用之前，我們先來看同步調用的定義；同步就是整個處理過程順序執行，當各個過程都執行完畢，並返回結果。 異步調用則是只是發送了調用的指令，調用者無需等待被調用的方法完全執行完畢；而是繼續執行下面的流程。

     例如， 在某個調用中，需要順序調用 A, B, C三個過程方法；如他們都是同步調用，則需要將他們都順序執行完畢之后，方算作過程執行完畢； 如B為一個異步的調用方法，則在執行完A之后，調用B，並不等待B完成，而是執行開始調用C，待C執行完畢之后，就意味着這個過程執行完畢了。

2.  常規的異步調用處理方式

    在Java中，一般在處理類似的場景之時，都是基於創建獨立的線程去完成相應的異步調用邏輯，通過主線程和不同的線程之間的執行流程，從而在啟動獨立的線程之后，主線程繼續執行而不會產生停滯等待的情況。

3. @Async介紹

   在Spring中，基於@Async標注的方法，稱之為異步方法；這些方法將在執行的時候，將會在獨立的線程中被執行，調用者無需等待它的完成，即可繼續其他的操作。

注意： @Async所修飾的函數不要定義為static類型，這樣異步調用不會生效，會報如下錯誤：

從異常信息JedisConnectionException: Could not get a resource from the pool來看，我們很容易的可以想到，在應用關閉的時候異步任務還在執行，由於Redis連接池先銷毀了，導致異步任務中要訪問Redis的操作就報了上面的錯。所以，我們得出結論，上面的實現方式在應用關閉的時候是不優雅的，那么我們要怎么做呢？如下設置：
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);

如何在Spring中啟用@Async

1、基於Java配置的啟用方式：

@Configuration  
@EnableAsync  
public class SpringAsyncConfig { ... }  

springboot中的配置是：

@EnableSwagger2
@EnableAsync
@EnableTransactionManagement
public class SettlementApplication {

    public static void main(String[] args) {
        
        SpringApplication.run(SettlementApplication.class, args);
    }
    
    @Bean
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
    
}

2、基於XML配置文件的啟用方式，配置如下：

<task:executor id="myexecutor" pool-size="5"  />  
<task:annotation-driven executor="myexecutor"/>  

 以上就是兩種定義的方式。

4. 基於@Async無返回值調用

    示例如下：

@Async  //標注使用  
public void asyncMethodWithVoidReturnType() {  
    System.out.println("Execute method asynchronously. "  
      + Thread.currentThread().getName());  
}  

  使用的方式非常簡單，一個標注即可解決所有的問題。

5. 基於@Async返回值的調用

   示例如下：

@Async  
public Future<String> asyncMethodWithReturnType() {  
    System.out.println("Execute method asynchronously - "  
      + Thread.currentThread().getName());  
    try {  
        Thread.sleep(5000);  
        return new AsyncResult<String>("hello world !!!!");  
    } catch (InterruptedException e) {  
        //  
    }  
   
    return null;  
}   

 

   以上示例可以發現，返回的數據類型為Future類型，其為一個接口。具體的結果類型為AsyncResult,這個是需要注意的地方。

   調用返回結果的異步方法示例：

public void testAsyncAnnotationForMethodsWithReturnType()  
   throws InterruptedException, ExecutionException {  
    System.out.println("Invoking an asynchronous method. "  
      + Thread.currentThread().getName());  
    Future<String> future = asyncAnnotationExample.asyncMethodWithReturnType();  
   
    while (true) {  ///這里使用了循環判斷，等待獲取結果信息  
        if (future.isDone()) {  //判斷是否執行完畢  
            System.out.println("Result from asynchronous process - " + future.get());  
            break;  
        }  
        System.out.println("Continue doing something else. ");  
        Thread.sleep(1000);  
    }  
}  

  分析： 這些獲取異步方法的結果信息，是通過不停的檢查Future的狀態來獲取當前的異步方法是否執行完畢來實現的。

6. 基於@Async調用中的異常處理機制

在異步方法中，如果出現異常，對於調用者caller而言，是無法感知的。

看一個示例：

A、異步方法類：

/**
 * @author duanxz
 * 2018年8月2日 下午4:36:49
 */
@Component
public class AsyncServiceTest {

    @Async
    public void test() {
        System.out.println("AsyncServiceTest.test()");
        throw new IllegalArgumentException("sssssssssssssssssss");
    }
}

B、業務調用類

        //...
        test.test();
        System.out.println("channelUploadFileList()" + LocalDateTime.now());
        MLogModel model = new MLogModel();
       //...

看結果：

如果確實需要進行異常處理，則按照如下方法來進行處理：

    1.  自定義實現AsyncTaskExecutor的任務執行器，在這里定義處理具體異常的邏輯和方式。

    2.  配置由自定義的TaskExecutor替代內置的任務執行器

    示例步驟1，自定義的TaskExecutor

public class ExceptionHandlingAsyncTaskExecutor implements AsyncTaskExecutor {  
    private AsyncTaskExecutor executor;  
    public ExceptionHandlingAsyncTaskExecutor(AsyncTaskExecutor executor) {  
        this.executor = executor;  
     }  
      ////用獨立的線程來包裝，@Async其本質就是如此  
    public void execute(Runnable task) {       
      executor.execute(createWrappedRunnable(task));  
    }  
    public void execute(Runnable task, long startTimeout) {  
        /用獨立的線程來包裝，@Async其本質就是如此  
       executor.execute(createWrappedRunnable(task), startTimeout);           
    }   
    public Future submit(Runnable task) { return executor.submit(createWrappedRunnable(task));  
       //用獨立的線程來包裝，@Async其本質就是如此。  
    }   
    public Future submit(final Callable task) {  
      //用獨立的線程來包裝，@Async其本質就是如此。  
       return executor.submit(createCallable(task));   
    }   
      
    private Callable createCallable(final Callable task) {   
        return new Callable() {   
            public T call() throws Exception {   
                 try {   
                     return task.call();   
                 } catch (Exception ex) {   
                     handle(ex);   
                     throw ex;   
                   }   
                 }   
        };   
    }  
  
    private Runnable createWrappedRunnable(final Runnable task) {   
         return new Runnable() {   
             public void run() {   
                 try {  
                     task.run();   
                  } catch (Exception ex) {   
                     handle(ex);   
                   }   
            }  
        };   
    }   
    private void handle(Exception ex) {  
      //具體的異常邏輯處理的地方  
      System.err.println("Error during @Async execution: " + ex);  
    }  
} 

 分析： 可以發現其是實現了AsyncTaskExecutor, 用獨立的線程來執行具體的每個方法操作。在createCallable和createWrapperRunnable中，定義了異常的處理方式和機制。

handle()就是未來我們需要關注的異常處理的地方。

配置文件中的內容：

    <task:annotation-driven executor="exceptionHandlingTaskExecutor" scheduler="defaultTaskScheduler" />  
    <bean id="exceptionHandlingTaskExecutor" class="nl.jborsje.blog.examples.ExceptionHandlingAsyncTaskExecutor">  
        <constructor-arg ref="defaultTaskExecutor" />  
    </bean>  
    <task:executor id="defaultTaskExecutor" pool-size="5" />  
    <task:scheduler id="defaultTaskScheduler" pool-size="1" />  

分析： 這里的配置使用自定義的taskExecutor來替代缺省的TaskExecutor。

或者（先看看Aysnc的源碼）：

public interface AsyncConfigurer {  
    Executor getAsyncExecutor();  
    AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler();  
} 

AsyncConfigurerSupport是AsyncConfigurer接口的實現但里邊什么也沒做。

public class AsyncConfigurerSupport implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        return null;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }

}

 

@Configuration  
@EnableAsync  
class SpringAsyncConfigurer extends AsyncConfigurerSupport {  
  
    @Bean  
    public ThreadPoolTaskExecutor asyncExecutor() {  
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();  
        threadPool.setCorePoolSize(3);  
        threadPool.setMaxPoolSize(3);  
        threadPool.setWaitForTasksToCompleteOnShutdown(true);  
        threadPool.setAwaitTerminationSeconds(60 * 15);  
        return threadPool;  
    }  
  
    @Override  
    public Executor getAsyncExecutor() {  
        return asyncExecutor;  
    }  
} 

可以自己實現AsyncConfigurer接口處理異常。

    @Configuration  
    @EnableAsync  
    public class SpringAsyncConfigurer implements AsyncConfigurer {  
           
        @Override  
        public Executor getAsyncExecutor() {  
            return new ThreadPoolTaskExecutor();  
        }  
      
        @Override  
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {  
            return new CustomAsyncExceptionHandler();  
        }  
      
    }  

異常處理類：

    public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {  
       
        @Override  
        public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {  
            System.out.println("Exception message - " + throwable.getMessage());  
            System.out.println("Method name - " + method.getName());  
            for (Object param : obj) {  
                System.out.println("Parameter value - " + param);  
            }  
        }  
           
    }  

 

7. @Async調用中的事務處理機制

    在@Async標注的方法，同時也適用了@Transactional進行了標注；在其調用數據庫操作之時，將無法產生事務管理的控制，原因就在於其是基於異步處理的操作。

     那該如何給這些操作添加事務管理呢？可以將需要事務管理操作的方法放置到異步方法內部，在內部被調用的方法上添加@Transactional.

    例如：  方法A，使用了@Async/@Transactional來標注，但是無法產生事務控制的目的。

          方法B，使用了@Async來標注，  B中調用了C、D，C/D分別使用@Transactional做了標注，則可實現事務控制的目的。

 

8. 異步線程池的定義

8.1、一個線程池

 

    @Configuration  
    @EnableAsync  
    public class SpringAsyncConfig {  
        @Bean  
        public AsyncTaskExecutor taskExecutor() {  
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();  
            executor.setMaxPoolSize(10);  
            return executor;  
        }  
    }  

8.2、多個線程池

    @Configuration  
    @EnableAsync  
    public class SpringAsyncConfig {  
           
        @Bean(name = "threadPoolTaskExecutor1")  
        public Executor threadPoolTaskExecutor() {  
            return new ThreadPoolTaskExecutor();  
        }  
           
        @Bean(name = "threadPoolTaskExecutor2")  
        public Executor threadPoolTaskExecutor() {  
            return new ThreadPoolTaskExecutor();  
        }  
      
    }  

調用

@Async("threadPoolTaskExecutor1")  
public void asyncMethodWithConfiguredExecutor() {  
    System.out.println("Execute method with configured executor - "  
      + Thread.currentThread().getName());  
}

 

9. 總結

     通過以上的描述，應該對@Async使用的方法和注意事項了。

 

轉自：Spring中@Async用法總結