[[334572]]

 1、本文內(nèi)容

詳解 @EnableAsync & @Async，主要分下面幾個(gè)點(diǎn)進(jìn)行介紹。

1. 作用
2. 用法
3. 獲取異步執(zhí)行結(jié)果
4. 自定義異步執(zhí)行的線程池
5. 自定義異常處理
6. 線程隔離
7. 源碼 & 原理

2、作用

spring容器中實(shí)現(xiàn)bean方法的異步調(diào)用。

比如有個(gè)logService的bean，logservice中有個(gè)log方法用來記錄日志，當(dāng)調(diào)用logService.log(msg)的時(shí)候，希望異步執(zhí)行，那么可以通過@EnableAsync & @Async來實(shí)現(xiàn)。

3、用法

2步

1. 需要異步執(zhí)行的方法上面使用@Async注解標(biāo)注，若bean中所有的方法都需要異步執(zhí)行，可以直接將@Async加載類上。
2. 將@EnableAsync添加在spring配置類上，此時(shí)@Async注解才會(huì)起效。

常見2種用法

1. 無返回值的
2. 可以獲取返回值的

4、無返回值的

用法

方法返回值不是Future類型的，被執(zhí)行時(shí)，會(huì)立即返回，并且無法獲取方法返回值，如：

@Async 
public void log(String msg) throws InterruptedException { 
    System.out.println("開始記錄日志," + System.currentTimeMillis()); 
    //模擬耗時(shí)2秒 
    TimeUnit.SECONDS.sleep(2); 
    System.out.println("日志記錄完畢," + System.currentTimeMillis()); 
} 

案例

實(shí)現(xiàn)日志異步記錄的功能。

LogService.log方法用來異步記錄日志，需要使用@Async標(biāo)注

package com.javacode2018.async.demo1; 
 
import org.springframework.scheduling.annotation.Async; 
import org.springframework.stereotype.Component; 
 
import java.util.concurrent.TimeUnit; 
 
@Component 
public class LogService { 
    @Async 
    public void log(String msg) throws InterruptedException { 
        System.out.println(Thread.currentThread() + "開始記錄日志," + System.currentTimeMillis()); 
        //模擬耗時(shí)2秒 
        TimeUnit.SECONDS.sleep(2); 
        System.out.println(Thread.currentThread() + "日志記錄完畢," + System.currentTimeMillis()); 
    } 
} 

來個(gè)spring配置類，需要加上@EnableAsync開啟bean方法的異步調(diào)用.

package com.javacode2018.async.demo1; 
 
import org.springframework.context.annotation.ComponentScan; 
import org.springframework.context.annotation.EnableAspectJAutoProxy; 
import org.springframework.scheduling.annotation.EnableAsync; 
 
@ComponentScan 
@EnableAsync 
public class MainConfig1 { 
} 

測試代碼

package com.javacode2018.async; 
 
import com.javacode2018.async.demo1.LogService; 
import com.javacode2018.async.demo1.MainConfig1; 
import org.junit.Test; 
import org.springframework.context.annotation.AnnotationConfigApplicationContext; 
 
import java.util.concurrent.TimeUnit; 
 
public class AsyncTest { 
 
    @Test 
    public void test1() throws InterruptedException { 
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); 
        context.register(MainConfig1.class); 
        context.refresh(); 
        LogService logService = context.getBean(LogService.class); 
        System.out.println(Thread.currentThread() + " logService.log start," + System.currentTimeMillis()); 
        logService.log("異步執(zhí)行方法!"); 
        System.out.println(Thread.currentThread() + " logService.log end," + System.currentTimeMillis()); 
 
        //休眠一下，防止@Test退出 
        TimeUnit.SECONDS.sleep(3); 
    } 
 
} 

運(yùn)行輸出

Thread[main,5,main] logService.log start,1595223990417 
Thread[main,5,main] logService.log end,1595223990432 
Thread[SimpleAsyncTaskExecutor-1,5,main]開始記錄日志,1595223990443 
Thread[SimpleAsyncTaskExecutor-1,5,main]日志記錄完畢,1595223992443 

前2行輸出，可以看出logService.log立即就返回了，后面2行來自于log方法，相差2秒左右。

前面2行在主線程中執(zhí)行，后面2行在異步線程中執(zhí)行。

5、獲取異步返回值

用法

若需取異步執(zhí)行結(jié)果，方法返回值必須為Future類型，使用spring提供的靜態(tài)方法org.springframework.scheduling.annotation.AsyncResult#forValue創(chuàng)建返回值，如：

public Future<String> getGoodsInfo(long goodsId) throws InterruptedException { 
    return AsyncResult.forValue(String.format("商品%s基本信息!", goodsId)); 
} 

案例

場景：電商中商品詳情頁通常會(huì)有很多信息：商品基本信息、商品描述信息、商品評(píng)論信息，通過3個(gè)方法來或者這幾個(gè)信息。

這3個(gè)方法之間無關(guān)聯(lián)，所以可以采用異步的方式并行獲取，提升效率。

下面是商品服務(wù)，內(nèi)部3個(gè)方法都需要異步，所以直接在類上使用@Async標(biāo)注了，每個(gè)方法內(nèi)部休眠500毫秒，模擬一下耗時(shí)操作。

package com.javacode2018.async.demo2; 
 
import org.springframework.scheduling.annotation.Async; 
import org.springframework.scheduling.annotation.AsyncResult; 
import org.springframework.stereotype.Component; 
 
import java.util.Arrays; 
import java.util.List; 
import java.util.concurrent.Future; 
import java.util.concurrent.TimeUnit; 
 
@Async 
@Component 
public class GoodsService { 
    //模擬獲取商品基本信息，內(nèi)部耗時(shí)500毫秒 
    public Future<String> getGoodsInfo(long goodsId) throws InterruptedException { 
        TimeUnit.MILLISECONDS.sleep(500); 
        return AsyncResult.forValue(String.format("商品%s基本信息!", goodsId)); 
    } 
 
    //模擬獲取商品描述信息，內(nèi)部耗時(shí)500毫秒 
    public Future<String> getGoodsDesc(long goodsId) throws InterruptedException { 
        TimeUnit.MILLISECONDS.sleep(500); 
        return AsyncResult.forValue(String.format("商品%s描述信息!", goodsId)); 
    } 
 
    //模擬獲取商品評(píng)論信息列表，內(nèi)部耗時(shí)500毫秒 
    public Future<List<String>> getGoodsComments(long goodsId) throws InterruptedException { 
        TimeUnit.MILLISECONDS.sleep(500); 
        List<String> comments = Arrays.asList("評(píng)論1", "評(píng)論2"); 
        return AsyncResult.forValue(comments); 
    } 
} 

來個(gè)spring配置類，需要加上@EnableAsync開啟bean方法的異步調(diào)用.

package com.javacode2018.async.demo2; 
 
import org.springframework.context.annotation.ComponentScan; 
import org.springframework.scheduling.annotation.EnableAsync; 
 
@ComponentScan 
@EnableAsync 
public class MainConfig2 { 
} 

測試代碼

@Test 
public void test2() throws InterruptedException, ExecutionException { 
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); 
    context.register(MainConfig2.class); 
    context.refresh(); 
    GoodsService goodsService = context.getBean(GoodsService.class); 
 
    long starTime = System.currentTimeMillis(); 
    System.out.println("開始獲取商品的各種信息"); 
 
    long goodsId = 1L; 
    Future<String> goodsInfoFuture = goodsService.getGoodsInfo(goodsId); 
    Future<String> goodsDescFuture = goodsService.getGoodsDesc(goodsId); 
    Future<List<String>> goodsCommentsFuture = goodsService.getGoodsComments(goodsId); 
 
    System.out.println(goodsInfoFuture.get()); 
    System.out.println(goodsDescFuture.get()); 
    System.out.println(goodsCommentsFuture.get()); 
 
    System.out.println("商品信息獲取完畢,總耗時(shí)(ms)：" + (System.currentTimeMillis() - starTime)); 
 
    //休眠一下，防止@Test退出 
    TimeUnit.SECONDS.sleep(3); 
} 

運(yùn)行輸出

開始獲取商品的各種信息 
商品1基本信息! 
商品1描述信息! 
[評(píng)論1, 評(píng)論2] 
商品信息獲取完畢,總耗時(shí)(ms)：525 

3個(gè)方法總計(jì)耗時(shí)500毫秒左右。

如果不采用異步的方式，3個(gè)方法會(huì)同步執(zhí)行，耗時(shí)差不多1.5秒，來試試，將GoodsService上的@Async去掉，然后再次執(zhí)行測試案例，輸出

開始獲取商品的各種信息 
商品1基本信息! 
商品1描述信息! 
[評(píng)論1, 評(píng)論2] 
商品信息獲取完畢,總耗時(shí)(ms)：1503 

這個(gè)案例大家可以借鑒一下，按照這個(gè)思路可以去優(yōu)化一下你們的代碼，方法之間無關(guān)聯(lián)的可以采用異步的方式，并行去獲取，最終耗時(shí)為最長的那個(gè)方法，整體相對(duì)于同步的方式性能提升不少。

6、自定義異步執(zhí)行的線程池

默認(rèn)情況下，@EnableAsync使用內(nèi)置的線程池來異步調(diào)用方法，不過我們也可以自定義異步執(zhí)行任務(wù)的線程池。

有2種方式來自定義異步處理的線程池

方式1

在spring容器中定義一個(gè)線程池類型的bean，bean名稱必須是taskExecutor

@Bean 
public Executor taskExecutor() { 
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
    executor.setCorePoolSize(10); 
    executor.setMaxPoolSize(100); 
    executor.setThreadNamePrefix("my-thread-"); 
    return executor; 
} 

方式2

定義一個(gè)bean，實(shí)現(xiàn)AsyncConfigurer接口中的getAsyncExecutor方法，這個(gè)方法需要返回自定義的線程池，案例代碼：

package com.javacode2018.async.demo3; 
 
import com.javacode2018.async.demo1.LogService; 
import org.springframework.beans.factory.annotation.Qualifier; 
import org.springframework.context.annotation.Bean; 
import org.springframework.lang.Nullable; 
import org.springframework.scheduling.annotation.AsyncConfigurer; 
import org.springframework.scheduling.annotation.EnableAsync; 
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; 
 
import java.util.concurrent.Executor; 
 
@EnableAsync 
public class MainConfig3 { 
 
    @Bean 
    public LogService logService() { 
        return new LogService(); 
    } 
 
    /** 
     * 定義一個(gè)AsyncConfigurer類型的bean，實(shí)現(xiàn)getAsyncExecutor方法，返回自定義的線程池 
     * 
     * @param executor 
     * @return 
     */ 
    @Bean 
    public AsyncConfigurer asyncConfigurer(@Qualifier("logExecutors") Executor executor) { 
        return new AsyncConfigurer() { 
            @Nullable 
            @Override 
            public Executor getAsyncExecutor() { 
                return executor; 
            } 
        }; 
    } 
 
    /** 
     * 定義一個(gè)線程池，用來異步處理日志方法調(diào)用 
     * 
     * @return 
     */ 
    @Bean 
    public Executor logExecutors() { 
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
        executor.setCorePoolSize(10); 
        executor.setMaxPoolSize(100); 
        //線程名稱前綴 
        executor.setThreadNamePrefix("log-thread-"); //@1 
        return executor; 
    } 
 
} 

@1自定義的線程池中線程名稱前綴為log-thread-，運(yùn)行下面測試代碼

@Test 
public void test3() throws InterruptedException { 
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); 
    context.register(MainConfig3.class); 
    context.refresh(); 
    LogService logService = context.getBean(LogService.class); 
    System.out.println(Thread.currentThread() + " logService.log start," + System.currentTimeMillis()); 
    logService.log("異步執(zhí)行方法!"); 
    System.out.println(Thread.currentThread() + " logService.log end," + System.currentTimeMillis()); 
 
    //休眠一下，防止@Test退出 
    TimeUnit.SECONDS.sleep(3); 
} 

輸出

Thread[main,5,main] logService.log start,1595228732914 
Thread[main,5,main] logService.log end,1595228732921 
Thread[log-thread-1,5,main]開始記錄日志,1595228732930 
Thread[log-thread-1,5,main]日志記錄完畢,1595228734931 

最后2行日志中線程名稱是log-thread-，正是我們自定義線程池中的線程。

7、自定義異常處理

異步方法若發(fā)生了異常，我們?nèi)绾潍@取異常信息呢？此時(shí)可以通過自定義異常處理來解決。

異常處理分2種情況

1. 當(dāng)返回值是Future的時(shí)候，方法內(nèi)部有異常的時(shí)候，異常會(huì)向外拋出，可以對(duì)Future.get采用try..catch來捕獲異常
2. 當(dāng)返回值不是Future的時(shí)候，可以自定義一個(gè)bean，實(shí)現(xiàn)AsyncConfigurer接口中的getAsyncUncaughtExceptionHandler方法，返回自定義的異常處理器

情況1：返回值為Future類型

用法

通過try..catch來捕獲異常，如下

try { 
    Future<String> future = logService.mockException(); 
    System.out.println(future.get()); 
} catch (ExecutionException e) { 
    System.out.println("捕獲 ExecutionException 異常"); 
    //通過e.getCause獲取實(shí)際的異常信息 
    e.getCause().printStackTrace(); 
} catch (InterruptedException e) { 
    e.printStackTrace(); 
} 

案例

LogService中添加一個(gè)方法，返回值為Future，內(nèi)部拋出一個(gè)異常，如下：

@Async 
public Future<String> mockException() { 
    //模擬拋出一個(gè)異常 
    throw new IllegalArgumentException("參數(shù)有誤!"); 
} 

測試代碼如下

@Test 
public void test5() throws InterruptedException { 
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); 
    context.register(MainConfig1.class); 
    context.refresh(); 
    LogService logService = context.getBean(LogService.class); 
    try { 
        Future<String> future = logService.mockException(); 
        System.out.println(future.get()); 
    } catch (ExecutionException e) { 
        System.out.println("捕獲 ExecutionException 異常"); 
        //通過e.getCause獲取實(shí)際的異常信息 
        e.getCause().printStackTrace(); 
    } catch (InterruptedException e) { 
        e.printStackTrace(); 
    } 
    //休眠一下，防止@Test退出 
    TimeUnit.SECONDS.sleep(3); 
} 

運(yùn)行輸出

java.lang.IllegalArgumentException: 參數(shù)有誤! 
捕獲 ExecutionException 異常 
 at com.javacode2018.async.demo1.LogService.mockException(LogService.java:23) 
 at com.javacode2018.async.demo1.LogService$$FastClassBySpringCGLIB$$32a28430.invoke(<generated>) 
 at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) 

情況2：無返回值異常處理

用法

當(dāng)返回值不是Future的時(shí)候，可以自定義一個(gè)bean，實(shí)現(xiàn)AsyncConfigurer接口中的getAsyncUncaughtExceptionHandler方法，返回自定義的異常處理器，當(dāng)目標(biāo)方法執(zhí)行過程中拋出異常的時(shí)候，此時(shí)會(huì)自動(dòng)回調(diào)AsyncUncaughtExceptionHandler#handleUncaughtException這個(gè)方法，可以在這個(gè)方法中處理異常，如下：

@Bean 
public AsyncConfigurer asyncConfigurer() { 
    return new AsyncConfigurer() { 
        @Nullable 
        @Override 
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { 
            return new AsyncUncaughtExceptionHandler() { 
                @Override 
                public void handleUncaughtException(Throwable ex, Method method, Object... params) { 
                    //當(dāng)目標(biāo)方法執(zhí)行過程中拋出異常的時(shí)候，此時(shí)會(huì)自動(dòng)回調(diào)這個(gè)方法，可以在這個(gè)方法中處理異常 
                } 
            }; 
        } 
    }; 
} 

案例

LogService中添加一個(gè)方法，內(nèi)部拋出一個(gè)異常，如下：

@Async 
public void mockNoReturnException() { 
    //模擬拋出一個(gè)異常 
    throw new IllegalArgumentException("無返回值的異常!"); 
} 

來個(gè)spring配置類，通過AsyncConfigurer來自定義異常處理器AsyncUncaughtExceptionHandler

package com.javacode2018.async.demo4; 
 
import com.javacode2018.async.demo1.LogService; 
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler; 
import org.springframework.context.annotation.Bean; 
import org.springframework.lang.Nullable; 
import org.springframework.scheduling.annotation.AsyncConfigurer; 
import org.springframework.scheduling.annotation.EnableAsync; 
 
import java.lang.reflect.Method; 
import java.util.Arrays; 
 
@EnableAsync 
public class MainConfig4 { 
 
    @Bean 
    public LogService logService() { 
        return new LogService(); 
    } 
 
    @Bean 
    public AsyncConfigurer asyncConfigurer() { 
        return new AsyncConfigurer() { 
            @Nullable 
            @Override 
            public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { 
                return new AsyncUncaughtExceptionHandler() { 
                    @Override 
                    public void handleUncaughtException(Throwable ex, Method method, Object... params) { 
                        String msg = String.format("方法[%s],參數(shù)[%s],發(fā)送異常了，異常詳細(xì)信息:", method, Arrays.asList(params)); 
                        System.out.println(msg); 
                        ex.printStackTrace(); 
                    } 
                }; 
            } 
        }; 
    } 
 
} 

運(yùn)行輸出

方法[public void com.javacode2018.async.demo1.LogService.mockNoReturnException()],參數(shù)[[]],發(fā)送異常了，異常詳細(xì)信息: 
java.lang.IllegalArgumentException: 無返回值的異常! 
 at com.javacode2018.async.demo1.LogService.mockNoReturnException(LogService.java:29) 
 at com.javacode2018.async.demo1.LogService$$FastClassBySpringCGLIB$$32a28430.invoke(<generated>) 
 at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) 

8、線程池隔離

什么是線程池隔離？

一個(gè)系統(tǒng)中可能有很多業(yè)務(wù)，比如充值服務(wù)、提現(xiàn)服務(wù)或者其他服務(wù)，這些服務(wù)中都有一些方法需要異步執(zhí)行，默認(rèn)情況下他們會(huì)使用同一個(gè)線程池去執(zhí)行，如果有一個(gè)業(yè)務(wù)量比較大，占用了線程池中的大量線程，此時(shí)會(huì)導(dǎo)致其他業(yè)務(wù)的方法無法執(zhí)行，那么我們可以采用線程隔離的方式，對(duì)不同的業(yè)務(wù)使用不同的線程池，相互隔離，互不影響。

@Async注解有個(gè)value參數(shù)，用來指定線程池的bean名稱，方法運(yùn)行的時(shí)候，就會(huì)采用指定的線程池來執(zhí)行目標(biāo)方法。

使用步驟

1. 在spring容器中，自定義線程池相關(guān)的bean
2. @Async("線程池bean名稱")

案例

模擬2個(gè)業(yè)務(wù)：異步充值、異步提現(xiàn)；2個(gè)業(yè)務(wù)都采用獨(dú)立的線程池來異步執(zhí)行，互不影響。

異步充值服務(wù)

package com.javacode2018.async.demo5; 
 
import org.springframework.scheduling.annotation.Async; 
import org.springframework.stereotype.Component; 
 
@Component 
public class RechargeService { 
    //模擬異步充值 
    @Async(MainConfig5.RECHARGE_EXECUTORS_BEAN_NAME) 
    public void recharge() { 
        System.out.println(Thread.currentThread() + "模擬異步充值"); 
    } 
} 

異步提現(xiàn)服務(wù)

package com.javacode2018.async.demo5; 
 
import org.springframework.scheduling.annotation.Async; 
import org.springframework.stereotype.Component; 
 
@Component 
public class CashOutService { 
    //模擬異步提現(xiàn) 
    @Async(MainConfig5.CASHOUT_EXECUTORS_BEAN_NAME) 
    public void cashOut() { 
        System.out.println(Thread.currentThread() + "模擬異步提現(xiàn)"); 
    } 
} 

spring配置類

注意@0、@1、@2、@3、@4這幾個(gè)地方的代碼，采用線程池隔離的方式，注冊了2個(gè)線程池，分別用來處理上面的2個(gè)異步業(yè)務(wù)。

package com.javacode2018.async.demo5; 
 
import org.springframework.context.annotation.Bean; 
import org.springframework.context.annotation.ComponentScan; 
import org.springframework.scheduling.annotation.EnableAsync; 
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; 
 
import java.util.concurrent.Executor; 
 
@EnableAsync //@0：啟用方法異步調(diào)用 
@ComponentScan 
public class MainConfig5 { 
 
    //@1：值業(yè)務(wù)線程池bean名稱 
    public static final String RECHARGE_EXECUTORS_BEAN_NAME = "rechargeExecutors"; 
    //@2：提現(xiàn)業(yè)務(wù)線程池bean名稱 
    public static final String CASHOUT_EXECUTORS_BEAN_NAME = "cashOutExecutors"; 
 
    /** 
     * @3：充值的線程池，線程名稱以recharge-thread-開頭 
     * @return 
     */ 
    @Bean(RECHARGE_EXECUTORS_BEAN_NAME) 
    public Executor rechargeExecutors() { 
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
        executor.setCorePoolSize(10); 
        executor.setMaxPoolSize(100); 
        //線程名稱前綴 
        executor.setThreadNamePrefix("recharge-thread-"); 
        return executor; 
    } 
 
    /** 
     * @4: 充值的線程池，線程名稱以cashOut-thread-開頭 
     * 
     * @return 
     */ 
    @Bean(CASHOUT_EXECUTORS_BEAN_NAME) 
    public Executor cashOutExecutors() { 
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
        executor.setCorePoolSize(10); 
        executor.setMaxPoolSize(100); 
        //線程名稱前綴 
        executor.setThreadNamePrefix("cashOut-thread-"); 
        return executor; 
    } 
} 

測試代碼

@Test 
public void test7() throws InterruptedException { 
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); 
    context.register(MainConfig5.class); 
    context.refresh(); 
 
    RechargeService rechargeService = context.getBean(RechargeService.class); 
    rechargeService.recharge(); 
    CashOutService cashOutService = context.getBean(CashOutService.class); 
    cashOutService.cashOut(); 
 
    //休眠一下，防止@Test退出 
    TimeUnit.SECONDS.sleep(3); 
} 

運(yùn)行輸出

Thread[recharge-thread-1,5,main]模擬異步充值 
Thread[cashOut-thread-1,5,main]模擬異步提現(xiàn) 

輸出中可以看出2個(gè)業(yè)務(wù)使用的是不同的線程池執(zhí)行的。

9、源碼 & 原理

內(nèi)部使用aop實(shí)現(xiàn)的，@EnableAsync會(huì)引入一個(gè)bean后置處理器：AsyncAnnotationBeanPostProcessor，將其注冊到spring容器，這個(gè)bean后置處理器在所有bean創(chuàng)建過程中，判斷bean的類上是否有@Async注解或者類中是否有@Async標(biāo)注的方法，如果有，會(huì)通過aop給這個(gè)bean生成代理對(duì)象，會(huì)在代理對(duì)象中添加一個(gè)切面：org.springframework.scheduling.annotation.AsyncAnnotationAdvisor，這個(gè)切面中會(huì)引入一個(gè)攔截器：AnnotationAsyncExecutionInterceptor，方法異步調(diào)用的關(guān)鍵代碼就是在這個(gè)攔截器的invoke方法中實(shí)現(xiàn)的，可以去看一下。