前言

Servlet 3.0之前：每一次Http請(qǐng)求都由一個(gè)線程從頭到尾處理。UFo28資訊網(wǎng)——每日最新資訊28at.com

Servlet 3.0之后，提供了異步處理請(qǐng)求：可以先釋放容器分配給請(qǐng)求的線程與相關(guān)資源，減輕系統(tǒng)負(fù)擔(dān)，從而增加服務(wù)的吞吐量。UFo28資訊網(wǎng)——每日最新資訊28at.com

在springboot應(yīng)用中，可以有4種方式實(shí)現(xiàn)異步接口（至于ResponseBodyEmitter、SseEmitter、StreamingResponseBody，不在本文介紹內(nèi)，之后新寫(xiě)文章介紹）：UFo28資訊網(wǎng)——每日最新資訊28at.com

• AsyncContext
• Callable
• WebAsyncTask
• DeferredResult

第一中AsyncContext是Servlet層級(jí)的，比較原生的方式，本文不對(duì)此介紹（一般都不使用它，太麻煩了）。本文著重介紹后面三種方式。UFo28資訊網(wǎng)——每日最新資訊28at.com

UFo28資訊網(wǎng)——每日最新資訊28at.com

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

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

Controller中，返回一個(gè)java.util.concurrent.Callable包裝的任何值，都表示該接口是一個(gè)異步接口：UFo28資訊網(wǎng)——每日最新資訊28at.com

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

服務(wù)器端的異步處理對(duì)客戶端來(lái)說(shuō)是不可見(jiàn)的。例如，上述接口，最終返回的客戶端的是一個(gè)String，和同步接口中，直接返回String的效果是一樣的。UFo28資訊網(wǎng)——每日最新資訊28at.com

Callable 處理過(guò)程如下：UFo28資訊網(wǎng)——每日最新資訊28at.com

控制器返回一個(gè) Callable 。UFo28資訊網(wǎng)——每日最新資訊28at.com

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

Callable默認(rèn)使用SimpleAsyncTaskExecutor類(lèi)來(lái)執(zhí)行，這個(gè)類(lèi)非常簡(jiǎn)單而且沒(méi)有重用線程。在實(shí)踐中，需要使用AsyncTaskExecutor類(lèi)來(lái)對(duì)線程進(jìn)行配置。UFo28資訊網(wǎng)——每日最新資訊28at.com

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

Spring提供的WebAsyncTask是對(duì)Callable的包裝，提供了更強(qiáng)大的功能，比如：處理超時(shí)回調(diào)、錯(cuò)誤回調(diào)、完成回調(diào)等。本質(zhì)上，和Callable區(qū)別不大，但是由于它額外封裝了一些事件的回調(diào)，所有，通常都使用WebAsyncTask而不是Callable：UFo28資訊網(wǎng)——每日最新資訊28at.com

@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;}

這里額外提一下，WebAsyncTask可以配置一個(gè)超時(shí)時(shí)間，這里配置的超時(shí)時(shí)間比全局配置的超時(shí)時(shí)間優(yōu)先級(jí)都高（會(huì)覆蓋全局配置的超時(shí)時(shí)間）。UFo28資訊網(wǎng)——每日最新資訊28at.com

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

DeferredResult使用方式與Callable類(lèi)似，但在返回結(jié)果時(shí)不一樣，它返回的時(shí)實(shí)際結(jié)果可能沒(méi)有生成，實(shí)際的結(jié)果可能會(huì)在另外的線程里面設(shè)置到DeferredResult中去。UFo28資訊網(wǎng)——每日最新資訊28at.com

//定義一個(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;}

如果調(diào)用以上接口，會(huì)發(fā)現(xiàn)客戶端的請(qǐng)求一直是在pending狀態(tài)——等待后端響應(yīng)。這里，我簡(jiǎn)單的將該接口返回的DeferredResult對(duì)象存放在了一個(gè)Map集合中，實(shí)際應(yīng)用中可以設(shè)計(jì)一個(gè)對(duì)象管理器來(lái)統(tǒng)一管理這些個(gè)對(duì)象。UFo28資訊網(wǎng)——每日最新資訊28at.com

UFo28資訊網(wǎng)——每日最新資訊28at.com

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

這里我又寫(xiě)了一個(gè)接口，模擬:UFo28資訊網(wǎng)——每日最新資訊28at.com

@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";}

其他線程修改DeferredResult的值：首先是從之前存放DeferredResult的map中拿到DeferredResult的值，然后設(shè)置它的返回值。當(dāng)執(zhí)行deferredResult.setResult之后，可以看到之前pending狀態(tài)的接口完成了響應(yīng)，得到的結(jié)果，就是這里設(shè)置的值。UFo28資訊網(wǎng)——每日最新資訊28at.com

這里也額外說(shuō)下：在返回DeferredResult時(shí)也可以設(shè)置超時(shí)時(shí)間，這個(gè)時(shí)間的優(yōu)先級(jí)也是大于全局設(shè)置的。另外，判斷DeferredResult是否有效，只是一個(gè)簡(jiǎn)單的判斷，實(shí)際中判斷有效的并不一定是有效的（比如：客戶端取消了請(qǐng)求，服務(wù)端是不知道的），但是一般判斷為無(wú)效的，那肯定是無(wú)效了。UFo28資訊網(wǎng)——每日最新資訊28at.com

DeferredResult 處理過(guò)程如下：UFo28資訊網(wǎng)——每日最新資訊28at.com

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

提供一個(gè)線程池

異步請(qǐng)求，不會(huì)一直占用請(qǐng)求的主線程（tomcat容器中處理請(qǐng)求的線程），而是通過(guò)一個(gè)其他的線程來(lái)處理異步任務(wù)。也正是如此，在相同的最大請(qǐng)求數(shù)配置下，異步請(qǐng)求由于迅速的釋放了主線程，所以才能提高吞吐量。UFo28資訊網(wǎng)——每日最新資訊28at.com

這里提到一個(gè)其他線程，那么這個(gè)其他線程我們一般都不適用默認(rèn)的，都是根據(jù)自身情況提供一個(gè)線程池供異步請(qǐng)求使用：（我給的參數(shù)都是測(cè)試用的，實(shí)際中不可照搬）。UFo28資訊網(wǎng)——每日最新資訊28at.com

@Bean("mvcAsyncTaskExecutor")public AsyncTaskExecutor asyncTaskExecutor() {    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();    // 線程池維護(hù)線程的最少數(shù)量    // asyncServiceExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() + 1);    executor.setCorePoolSize(5);    // 線程池維護(hù)線程的最大數(shù)量    executor.setMaxPoolSize(10);    // 線程池所使用的緩沖隊(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());    // 線程全部結(jié)束才關(guā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;}

把這個(gè)線程池配置設(shè)置到異步請(qǐng)求配置中：UFo28資訊網(wǎng)——每日最新資訊28at.com

@Configurationpublic 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)求

異步請(qǐng)求能提高吞吐量，這個(gè)是建立在相同配置（這里的配置指的是：最大連接數(shù)、最大工作線程數(shù)）的情況下。因此并不是說(shuō)任何接口都可以使用異步請(qǐng)求。比如：一個(gè)請(qǐng)求是進(jìn)行大量的計(jì)算（總之就是在處理這個(gè)請(qǐng)求的業(yè)務(wù)方法時(shí)CPU是沒(méi)有休息的），這種情況使用異步請(qǐng)求就沒(méi)有多大意義了，因?yàn)檫@時(shí)的異步請(qǐng)求只是把一個(gè)任務(wù)從tomcat的工作線程搬到了另一個(gè)線程罷了。UFo28資訊網(wǎng)——每日最新資訊28at.com

直接調(diào)大最大工作線程數(shù)配置也能到達(dá)要求。所以，真正使用異步請(qǐng)求的場(chǎng)景應(yīng)該是該請(qǐng)求的業(yè)務(wù)代碼中，大量的時(shí)間CPU是休息的（比如：在業(yè)務(wù)代碼中請(qǐng)求其他系統(tǒng)的接口，在其他系統(tǒng)響應(yīng)之前，CPU是阻塞等待的），這個(gè)時(shí)候使用異步請(qǐng)求，就可以釋放tomcat的工作線程，讓釋放的工作線程可以處理其他的請(qǐng)求，從而提高吞吐量。UFo28資訊網(wǎng)——每日最新資訊28at.com

由于異步請(qǐng)求增加了更多的線程切換（同步請(qǐng)求是同一個(gè)工作線程一直處理），所以理論上會(huì)增加接口的耗時(shí)。但，這個(gè)耗時(shí)很短很短。UFo28資訊網(wǎng)——每日最新資訊28at.com

本文鏈接：http://m.www897cc.com/showinfo-26-112714-0.htmlSpringBoot異步接口實(shí)現(xiàn)：提高系統(tǒng)的吞吐量

聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問(wèn)題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com

上一篇： 時(shí)間序列結(jié)構(gòu)變化分析：Python實(shí)現(xiàn)時(shí)間序列變化點(diǎn)檢測(cè)

下一篇： flat() 和 flatMap() 有什么區(qū)別？