您好,登錄后才能下訂單哦!
這篇文章主要介紹“Future模式介紹以及SpringBoot異步編程實例分析”,在日常操作中,相信很多人在Future模式介紹以及SpringBoot異步編程實例分析問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”Future模式介紹以及SpringBoot異步編程實例分析”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
通過本文你可以了解到下面這些知識點:
Future 模式介紹以及核心思想
核心線程數(shù)、最大線程數(shù)的區(qū)別,隊列容量代表什么;
ThreadPoolTaskExecutor 飽和策略;
SpringBoot 異步編程實戰(zhàn),搞懂代碼的執(zhí)行邏輯。
Future 模式
異步編程在處理耗時操作以及多任務(wù)處理的場景下非常有用,我們可以更好的讓我們的系統(tǒng)利用好機器的 CPU 和 內(nèi)存,提高它們的利用率。多線程設(shè)計模式有很多種,F(xiàn)uture模式是多線程開發(fā)中非常常見的一種設(shè)計模式,本文也是基于這種模式來說明 SpringBoot 對于異步編程的知識。
實戰(zhàn)之前我先簡單介紹一下 Future 模式的核心思想 吧!。
Future 模式的核心思想是 異步調(diào)用 。當我們執(zhí)行一個方法時,假如這個方法中有多個耗時的任務(wù)需要同時去做,而且又不著急等待這個結(jié)果時可以讓客戶端立即返回然后,后臺慢慢去計算任務(wù)。當然你也可以選擇等這些任務(wù)都執(zhí)行完了,再返回給客戶端。這個在 Java 中都有很好的支持,我在后面的示例程序中會詳細對比這兩種方式的區(qū)別。
SpringBoot 異步編程實戰(zhàn)
如果我們需要在 SpringBoot 實現(xiàn)異步編程的話,通過 Spring 提供的兩個注解會讓這件事情變的非常簡單。
@EnableAsync:通過在配置類或者Main類上加@EnableAsync開啟對異步方法的支持。
@Async 可以作用在類上或者方法上,作用在類上代表這個類的所有方法都是異步方法。
1. 自定義 TaskExecutor
很多人對于 TaskExecutor 不是太了解,所以我們花一點篇幅先介紹一下這個東西。從名字就能看出它是任務(wù)的執(zhí)行者,它領(lǐng)導(dǎo)執(zhí)行著線程來處理任務(wù),就像司令官一樣,而我們的線程就好比一只只軍隊一樣,這些軍隊可以異步對敵人進行打擊?。
Spring 提供了TaskExecutor接口作為任務(wù)執(zhí)行者的抽象,它和java.util.concurrent包下的Executor接口很像。稍微不同的 TaskExecutor接口用到了 Java 8 的語法@FunctionalInterface聲明這個接口是一個函數(shù)式接口。
org.springframework.core.task.TaskExecutor @FunctionalInterface public interface TaskExecutor extends Executor { void execute(Runnable var1); }
如果沒有自定義Executor, Spring 將創(chuàng)建一個 SimpleAsyncTaskExecutor
并使用它。
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; /** @author shuang.kou */ @Configuration @EnableAsync public class AsyncConfig implements AsyncConfigurer { private static final int CORE_POOL_SIZE = 6; private static final int MAX_POOL_SIZE = 10; private static final int QUEUE_CAPACITY = 100; @Bean public Executor taskExecutor() { // Spring 默認配置是核心線程數(shù)大小為1,最大線程容量大小不受限制,隊列容量也不受限制。 ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 核心線程數(shù) executor.setCorePoolSize(CORE_POOL_SIZE); // 最大線程數(shù) executor.setMaxPoolSize(MAX_POOL_SIZE); // 隊列大小 executor.setQueueCapacity(QUEUE_CAPACITY); // 當最大池已滿時,此策略保證不會丟失任務(wù)請求,但是可能會影響應(yīng)用程序整體性能。 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setThreadNamePrefix("My ThreadPoolTaskExecutor-"); executor.initialize(); return executor; } }
ThreadPoolTaskExecutor 常見概念:
Core Pool Size : 核心線程數(shù)線程數(shù)定義了最小可以同時運行的線程數(shù)量。
Queue Capacity : 當新任務(wù)來的時候會先判斷當前運行的線程數(shù)量是否達到核心線程數(shù),如果達到的話,信任就會被存放在隊列中。
Maximum Pool Size : 當隊列中存放的任務(wù)達到隊列容量的時候,當前可以同時運行的線程數(shù)量變?yōu)樽畲缶€程數(shù)。
一般情況下不會將隊列大小設(shè)為:Integer.MAX_VALUE,也不會將核心線程數(shù)和最大線程數(shù)設(shè)為同樣的大小,這樣的話最大線程數(shù)的設(shè)置都沒什么意義了,你也無法確定當前 CPU 和內(nèi)存利用率具體情況如何。
如果隊列已滿并且當前同時運行的線程數(shù)達到最大線程數(shù)的時候,如果再有新任務(wù)過來會發(fā)生什么呢?
Spring 默認使用的是 ThreadPoolExecutor.AbortPolicy。在Spring的默認情況下,ThreadPoolExecutor 將拋出 RejectedExecutionException 來拒絕新來的任務(wù) ,這代表你將丟失對這個任務(wù)的處理。 對于可伸縮的應(yīng)用程序,建議使用 ThreadPoolExecutor.CallerRunsPolicy。當最大池被填滿時,此策略為我們提供可伸縮隊列。
ThreadPoolTaskExecutor 飽和策略定義:
如果當前同時運行的線程數(shù)量達到最大線程數(shù)量時,ThreadPoolTaskExecutor 定義一些策略:
ThreadPoolExecutor.AbortPolicy:拋出 RejectedExecutionException來拒絕新任務(wù)的處理。
ThreadPoolExecutor.CallerRunsPolicy:調(diào)用執(zhí)行自己的線程運行任務(wù)。您不會任務(wù)請求。但是這種策略會降低對于新任務(wù)提交速度,影響程序的整體性能。另外,這個策略喜歡增加隊列容量。如果您的應(yīng)用程序可以承受此延遲并且你不能任務(wù)丟棄任何一個任務(wù)請求的話,你可以選擇這個策略。
ThreadPoolExecutor.DiscardPolicy: 不處理新任務(wù),直接丟棄掉。
ThreadPoolExecutor.DiscardOldestPolicy: 此策略將丟棄最早的未處理的任務(wù)請求。
2. 編寫一個異步的方法
下面模擬一個查找對應(yīng)字符開頭電影的方法,我們給這個方法加上了@Async注解來告訴 Spring 它是一個異步的方法。另外,這個方法的返回值 CompletableFuture.completedFuture(results)這代表我們需要返回結(jié)果,也就是說程序必須把任務(wù)執(zhí)行完成之后再返回給用戶。
請留意completableFutureTask方法中的第一行打印日志這句代碼,后面分析程序中會用到,很重要!
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; /** @author shuang.kou */ @Service public class AsyncService { private static final Logger logger = LoggerFactory.getLogger(AsyncService.class); private List<String> movies = new ArrayList<>( Arrays.asList( "Forrest Gump", "Titanic", "Spirited Away", "The Shawshank Redemption", "Zootopia", "Farewell ", "Joker", "Crawl")); /** 示范使用:找到特定字符/字符串開頭的電影 */ @Async public CompletableFuture<List<String>> completableFutureTask(String start) { // 打印日志 logger.warn(Thread.currentThread().getName() + "start this task!"); // 找到特定字符/字符串開頭的電影 List<String> results = movies.stream().filter(movie -> movie.startsWith(start)).collect(Collectors.toList()); // 模擬這是一個耗時的任務(wù) try { Thread.sleep(1000L); } catch (InterruptedException e) { e.printStackTrace(); } //返回一個已經(jīng)用給定值完成的新的CompletableFuture。 return CompletableFuture.completedFuture(results); } }
3. 測試編寫的異步方法
/** @author shuang.kou */ @RestController @RequestMapping("/async") public class AsyncController { @Autowired AsyncService asyncService; @GetMapping("/movies") public String completableFutureTask() throws ExecutionException, InterruptedException { //開始時間 long start = System.currentTimeMillis(); // 開始執(zhí)行大量的異步任務(wù) List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C"); List<CompletableFuture<List<String>>> completableFutureList = words.stream() .map(word -> asyncService.completableFutureTask(word)) .collect(Collectors.toList()); // CompletableFuture.join()方法可以獲取他們的結(jié)果并將結(jié)果連接起來 List<List<String>> results = completableFutureList.stream().map(CompletableFuture::join).collect(Collectors.toList()); // 打印結(jié)果以及運行程序運行花費時間 System.out.println("Elapsed time: " + (System.currentTimeMillis() - start)); return results.toString(); } }
請求這個接口,控制臺打印出下面的內(nèi)容:
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-1] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-1start this task!
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-6] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-6start this task!
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-5] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-5start this task!
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-4] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-4start this task!
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-3] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-3start this task!
2019-10-01 13:50:17.007 WARN 18793 --- [lTaskExecutor-2] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-2start this task!
Elapsed time: 1010
首先我們可以看到處理所有任務(wù)花費的時間大概是 1 s。這與我們自定義的 ThreadPoolTaskExecutor 有關(guān),我們配置的核心線程數(shù)是 6 ,然后通過通過下面的代碼模擬分配了 6 個任務(wù)給系統(tǒng)執(zhí)行。這樣每個線程都會被分配到一個任務(wù),每個任務(wù)執(zhí)行花費時間是 1 s ,所以處理 6 個任務(wù)的總花費時間是 1 s。
List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C"); List<CompletableFuture<List<String>>> completableFutureList = words.stream() .map(word -> asyncService.completableFutureTask(word)) .collect(Collectors.toList());
你可以自己驗證一下,試著去把核心線程數(shù)的數(shù)量改為 3 ,再次請求這個接口你會發(fā)現(xiàn)處理所有任務(wù)花費的時間大概是 2 s。
另外,從上面的運行結(jié)果可以看出,當所有任務(wù)執(zhí)行完成之后才返回結(jié)果。這種情況對應(yīng)于我們需要返回結(jié)果給客戶端請求的情況下,假如我們不需要返回任務(wù)執(zhí)行結(jié)果給客戶端的話呢? 就比如我們上傳一個大文件到系統(tǒng),上傳之后只要大文件格式符合要求我們就上傳成功。普通情況下我們需要等待文件上傳完畢再返回給用戶消息,但是這樣會很慢。采用異步的話,當用戶上傳之后就立馬返回給用戶消息,然后系統(tǒng)再默默去處理上傳任務(wù)。這樣也會增加一點麻煩,因為文件可能會上傳失敗,所以系統(tǒng)也需要一點機制來補償這個問題,比如當上傳遇到問題的時候,發(fā)消息通知用戶。
下面會演示一下客戶端不需要返回結(jié)果的情況:
將completableFutureTask方法變?yōu)?void 類型
@Async public void completableFutureTask(String start) { ...... //這里可能是系統(tǒng)對任務(wù)執(zhí)行結(jié)果的處理,比如存入到數(shù)據(jù)庫等等...... //doSomeThingWithResults(results); }
Controller 代碼修改如下:
@GetMapping("/movies") public String completableFutureTask() throws ExecutionException, InterruptedException { // Start the clock long start = System.currentTimeMillis(); // Kick of multiple, asynchronous lookups List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C"); words.stream() .forEach(word -> asyncService.completableFutureTask(word)); // Wait until they are all done // Print results, including elapsed time System.out.println("Elapsed time: " + (System.currentTimeMillis() - start)); return "Done"; }
請求這個接口,控制臺打印出下面的內(nèi)容:
Elapsed time: 0
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-4] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-4start this task!
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-3] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-3start this task!
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-2] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-2start this task!
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-1] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-1start this task!
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-6] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-6start this task!
2019-10-01 14:02:44.052 WARN 19051 --- [lTaskExecutor-5] g.j.a.service.AsyncService : My ThreadPoolTaskExecutor-5start this task!
可以看到系統(tǒng)會直接返回給用戶結(jié)果,然后系統(tǒng)才真正開始執(zhí)行任務(wù)。
到此,關(guān)于“Future模式介紹以及SpringBoot異步編程實例分析”的學習就結(jié)束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續(xù)學習更多相關(guān)知識,請繼續(xù)關(guān)注億速云網(wǎng)站,小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。