SpringBoot異步接口實(shí)現(xiàn)：提高系統(tǒng)的吞吐量

來(lái)自：juejin.cn/post/7367186272849788962

前言

• AsyncContext
• Callable
• WebAsyncTask
• DeferredResult

特別說(shuō)明：服務(wù)端的異步或同步對(duì)于客戶(hù)端而言是不可見(jiàn)的。不會(huì)因?yàn)榉?wù)端使用了異步，接口的結(jié)果就和同步不一樣了。另外，對(duì)于單個(gè)請(qǐng)求而言，使用異步接口會(huì)導(dǎo)致響應(yīng)時(shí)間比同步大，但不特別明顯。具體后文分析。

基于Callable實(shí)現(xiàn)

   

    @GetMapping("/testCallAble")
    
public Callable<String> testCallAble() {
    
    return () -> {
    
        Thread.sleep(40000);
    
        return "hello";
    
    };
    
}
    

   

• Spring MVC 調(diào)用 request.startAsync() 并將 Callable 提交給 AsyncTaskExecutor 以在單獨(dú)的線(xiàn)程中進(jìn)行處理。
• 同時(shí)， DispatcherServlet 和所有過(guò)濾器退出 Servlet 容器線(xiàn)程，但response保持打開(kāi)狀態(tài)。
• 最終 Callable 產(chǎn)生結(jié)果，Spring MVC將請(qǐng)求分派回Servlet容器以完成處理。
• 再次調(diào)用 DispatcherServlet ，并使用 Callable 異步生成的返回值繼續(xù)處理。

基于WebAsyncTask實(shí)現(xiàn)

   

    @GetMapping("/webAsyncTask")
    
public WebAsyncTask<String> webAsyncTask() {
    
    WebAsyncTask<String> result = new WebAsyncTask<>(30003, () -> {
    
        return "success";
    
    });
    
    result.onTimeout(() -> {
    
        log.info("timeout callback");
    
        return "timeout callback";
    
    });
    
    result.onCompletion(() -> log.info("finish callback"));
    
    return result;
    
}
    

   

基于DeferredResult實(shí)現(xiàn)

   

    //定義一個(gè)全局的變量，用來(lái)存儲(chǔ)DeferredResult對(duì)象
    
private Map<String, DeferredResult<String>> deferredResultMap = new ConcurrentHashMap<>();
    

    
@GetMapping("/testDeferredResult")
    
public DeferredResult<String> testDeferredResult(){
    
    DeferredResult<String> deferredResult = new DeferredResult<>();
    
    deferredResultMap.put("test", deferredResult);
    
    return deferredResult;
    
}
    

   

注意：要考慮定時(shí)輪詢(xún)（或其他方式）這些對(duì)象，將已經(jīng)處理過(guò)或無(wú)效的DeferredResult對(duì)象清理掉（DeferredResult.isSetOrExpired方法可以判斷是否還有效），避免內(nèi)存泄露。

   

    @GetMapping("/testSetDeferredResult")
    
public String testSetDeferredResult() throws InterruptedException {
    
    DeferredResult<String> deferredResult = deferredResultMap.get("test");
    
    boolean flag = deferredResult.setResult("testSetDeferredResult");
    
    if(!flag){
    
        log.info("結(jié)果已經(jīng)被處理，此次操作無(wú)效");
    
    }
    
    return "ok";
    
}
    

   

• 控制器返回一個(gè) DeferredResult 并將其保存在可以訪(fǎng)問(wèn)的內(nèi)存隊(duì)列或列表中。
• Spring MVC 調(diào)用 request.startAsync() 。
• 同時(shí)，DispatcherServlet 和所有配置的過(guò)濾器退出請(qǐng)求處理線(xiàn)程，但響應(yīng)保持打開(kāi)狀態(tài)。
• 應(yīng)用程序從某個(gè)線(xiàn)程設(shè)置 DeferredResult ，Spring MVC 將請(qǐng)求分派回 Servlet 容器。
• 再次調(diào)用 DispatcherServlet ，并使用異步生成的返回值繼續(xù)處理。

提供一個(gè)線(xiàn)程池

   

    @Bean("mvcAsyncTaskExecutor")
    
public AsyncTaskExecutor asyncTaskExecutor() {
    
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    
    // 線(xiàn)程池維護(hù)線(xiàn)程的最少數(shù)量
    
    // asyncServiceExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() + 1);
    
    executor.setCorePoolSize(5);
    
    // 線(xiàn)程池維護(hù)線(xiàn)程的最大數(shù)量
    
    executor.setMaxPoolSize(10);
    
    // 線(xiàn)程池所使用的緩沖隊(duì)列
    
    executor.setQueueCapacity(10);
    
    //   asyncServiceExecutor.prefersShortLivedTasks();
    
    executor.setThreadNamePrefix("fyk-mvcAsyncTask-Thread-");
    
    asyncServiceExecutor.setBeanName("TaskId" + taskId);
    
    //  asyncServiceExecutor.setKeepAliveSeconds(20);
    
    //調(diào)用者執(zhí)行
    
    //   asyncServiceExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
    
    // 線(xiàn)程全部結(jié)束才關(guān)閉線(xiàn)程池
    
    executor.setWaitForTasksToCompleteOnShutdown(true);
    
    // 如果超過(guò)60s還沒(méi)有銷(xiāo)毀就強(qiáng)制銷(xiāo)毀，以確保應(yīng)用最后能夠被關(guān)閉，而不是阻塞住
    
    executor.setAwaitTerminationSeconds(30);
    
    executor.initialize();
    

    
    return executor;
    
}
    

   

   

    @Configuration
    
public class FykWebMvcConfigurer implements WebMvcConfigurer {
    

    
    @Autowired
    
    @Qualifier("mvcAsyncTaskExecutor")
    
    private AsyncTaskExecutor asyncTaskExecutor;
    

    
    @Override
    
    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
    
        //異步操作的超時(shí)時(shí)間，值為0或者更小，表示永不超時(shí)
    
        configurer.setDefaultTimeout(60001);
    
        configurer.setTaskExecutor(asyncTaskExecutor);
    
    }
    
}
    

   

什么時(shí)候使用異步請(qǐng)求

推薦閱讀：

船新 IDEA 2024.2 正式發(fā)布，新特性真香！

如何為開(kāi)放平臺(tái)設(shè)計(jì)一個(gè)安全好用的OpenApi

     

      程序員在線(xiàn)工具站：cxytools.com
     
推薦一個(gè)自己寫(xiě)的工具站：http://cxytools.com，專(zhuān)為程序員設(shè)計(jì)，包括時(shí)間日期、JSON處理、SQL格式化、隨機(jī)字符串生成、UUID生成、文本Hash...等功能，提升開(kāi)發(fā)效率。
     

      ?戳閱讀原文直達(dá)！                                  朕已閱