java8異步調(diào)用如何使用才是最好的方式
一、異步調(diào)用方式分析
今天在寫(xiě)代碼的時(shí)候，想要調(diào)用異步的操作，這里我是用的java8的流式異步調(diào)用，但是使用過(guò)程中呢，發(fā)現(xiàn)這個(gè)異步方式有兩個(gè)方法，如下所示：
區(qū)別是一個(gè) 需要指定線程池，一個(gè)不需要。
那么指定線程池有哪些好處呢？直觀的說(shuō)有以下兩點(diǎn)好處：
- 可以根據(jù)我們的服務(wù)器性能，通過(guò)池的管理更好的規(guī)劃我們的線程數(shù)。
- 可以對(duì)我們使用的線程自定義名稱，這里也是阿里java開(kāi)發(fā)規(guī)范所提到的。
1.1 java8異步調(diào)用默認(rèn)線程池方式
當(dāng)然常規(guī)使用默認(rèn)的也沒(méi)什么問(wèn)題。我們通過(guò)源碼分析下使用默認(rèn)線程池的過(guò)程。
public static CompletableFuture<Void> runAsync(Runnable runnable) {
return asyncRunStage(asyncPool, runnable);
}看下這個(gè)asyncPool是什么？
如下所示，useCommonPool如果為真，就使用ForkJoinPool.commonPool()，否則創(chuàng)建一個(gè)new ThreadPerTaskExecutor()：
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();看看useCommonPool 是什么？
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1); /**
* 公共池的目標(biāo)并行度級(jí)別
*/
public static int getCommonPoolParallelism() {
return commonParallelism;
}最終這個(gè)并行級(jí)別并沒(méi)有給出默認(rèn)值
static final int commonParallelism;
通過(guò)找到這個(gè)常量的調(diào)用，我們看看是如何進(jìn)行初始化的，在ForkJoinPool中有一個(gè)靜態(tài)代碼塊，啟動(dòng)時(shí)會(huì)對(duì)commonParallelism進(jìn)行初始化，我們只關(guān)注最后一句話就好了，：
// Unsafe mechanics
private static final sun.misc.Unsafe U;
private static final int ABASE;
private static final int ASHIFT;
private static final long CTL;
private static final long RUNSTATE;
private static final long STEALCOUNTER;
private static final long PARKBLOCKER;
private static final long QTOP;
private static final long QLOCK;
private static final long QSCANSTATE;
private static final long QPARKER;
private static final long QCURRENTSTEAL;
private static final long QCURRENTJOIN;
static {
// initialize field offsets for CAS etc
try {
U = sun.misc.Unsafe.getUnsafe();
Class<?> k = ForkJoinPool.class;
CTL = U.objectFieldOffset
(k.getDeclaredField("ctl"));
RUNSTATE = U.objectFieldOffset
(k.getDeclaredField("runState"));
STEALCOUNTER = U.objectFieldOffset
(k.getDeclaredField("stealCounter"));
Class<?> tk = Thread.class;
PARKBLOCKER = U.objectFieldOffset
(tk.getDeclaredField("parkBlocker"));
Class<?> wk = WorkQueue.class;
QTOP = U.objectFieldOffset
(wk.getDeclaredField("top"));
QLOCK = U.objectFieldOffset
(wk.getDeclaredField("qlock"));
QSCANSTATE = U.objectFieldOffset
(wk.getDeclaredField("scanState"));
QPARKER = U.objectFieldOffset
(wk.getDeclaredField("parker"));
QCURRENTSTEAL = U.objectFieldOffset
(wk.getDeclaredField("currentSteal"));
QCURRENTJOIN = U.objectFieldOffset
(wk.getDeclaredField("currentJoin"));
Class<?> ak = ForkJoinTask[].class;
ABASE = U.arrayBaseOffset(ak);
int scale = U.arrayIndexScale(ak);
if ((scale & (scale - 1)) != 0)
throw new Error("data type scale not a power of two");
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw new Error(e);
}
commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
defaultForkJoinWorkerThreadFactory =
new DefaultForkJoinWorkerThreadFactory();
modifyThreadPermission = new RuntimePermission("modifyThread");
common = java.security.AccessController.doPrivileged
(new java.security.PrivilegedAction<ForkJoinPool>() {
public ForkJoinPool run() { return makeCommonPool(); }});
// 即使線程被禁用也是1，至少是個(gè)1
int par = common.config & SMASK;
commonParallelism = par > 0 ? par : 1;
}如下所示，默認(rèn)是7：
所以接著下面的代碼看：
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);這里一定是返回true，證明當(dāng)前是并行的。
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();上面會(huì)返回一個(gè)大小是七的的默認(rèn)線程池
其實(shí)這個(gè)默認(rèn)值是當(dāng)前cpu的核心數(shù)，我的電腦是八核，在代碼中默認(rèn)會(huì)將核心數(shù)減一，所以顯示是七個(gè)線程。
if (parallelism < 0 && //默認(rèn)是1，小于核心數(shù)
(parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
parallelism = 1;
if (parallelism > MAX_CAP)
parallelism = MAX_CAP;下面我們寫(xiě)個(gè)main方法測(cè)試一下，10個(gè)線程，每個(gè)阻塞10秒，看結(jié)果：
public static void main(String[] args) {
// 創(chuàng)建10個(gè)任務(wù)，每個(gè)任務(wù)阻塞10秒
for (int i = 0; i < 10; i++) {
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(10000);
System.out.println(new Date() + ":" + Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}結(jié)果如下所示，前面七個(gè)任務(wù)先完成，另外三個(gè)任務(wù)被阻塞10秒后，才完成：
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-5 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-4 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-2 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-7 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-3 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-6 Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-1 ----------------------------------------------------------- Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-2 Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-5 Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-4
結(jié)論：當(dāng)我們使用默認(rèn)的線程池進(jìn)行異步調(diào)用時(shí)，如果異步任務(wù)是一個(gè)IO密集型，簡(jiǎn)單說(shuō)處理時(shí)間占用長(zhǎng)，將導(dǎo)致其他使用共享線程池的任務(wù)阻塞，造成系統(tǒng)性能下降甚至異常。甚至當(dāng)一部分調(diào)用接口時(shí)，如果接口超時(shí)，那么也會(huì)阻塞與超時(shí)時(shí)長(zhǎng)相同的時(shí)間；實(shí)際在計(jì)算密集的場(chǎng)景下使用是能提高性能的。
二、使用自定義的線程池
上面說(shuō)到如果是IO密集型的場(chǎng)景，在異步調(diào)用時(shí)還是使用自定義線程池比較好。
針對(duì)開(kāi)篇提到的兩個(gè)顯而易見(jiàn)的好處，此處新增一條：
- 可以根據(jù)我們的服務(wù)器性能，通過(guò)池的管理更好的規(guī)劃我們的線程數(shù)。
- 可以對(duì)我們使用的線程自定義名稱，這里也是阿里java開(kāi)發(fā)規(guī)范所提到的。
- 不會(huì)因?yàn)樽枞麑?dǎo)致使用共享線程池的其他線程阻塞甚至異常。
我們自定義下面的線程池：
/**
* @description： 全局通用線程池
* @author：weirx
* @date：2021/9/9 18:09
* @version：3.0
*/
@Slf4j
public class GlobalThreadPool {
/**
* 核心線程數(shù)
*/
public final static int CORE_POOL_SIZE = 10;
/**
* 最大線程數(shù)
*/
public final static int MAX_NUM_POOL_SIZE = 20;
/**
* 任務(wù)隊(duì)列大小
*/
public final static int BLOCKING_QUEUE_SIZE = 30;
/**
* 線程池實(shí)例
*/
private final static ThreadPoolExecutor instance = getInstance();
/**
* description: 初始化線程池
*
* @return: java.util.concurrent.ThreadPoolExecutor
* @author: weirx
* @time: 2021/9/10 9:49
*/
private synchronized static ThreadPoolExecutor getInstance() {
// 生成線程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_NUM_POOL_SIZE,
60,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(BLOCKING_QUEUE_SIZE),
new NamedThreadFactory("Thread-wjbgn-", false));
return executor;
}
private GlobalThreadPool() {
}
public static ThreadPoolExecutor getExecutor() {
return instance;
}
}調(diào)用：
public static void main(String[] args) {
// 創(chuàng)建10個(gè)任務(wù)，每個(gè)任務(wù)阻塞10秒
for (int i = 0; i < 10; i++) {
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(10000);
System.out.println(new Date() + ":" + Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
},GlobalThreadPool.getExecutor());
}
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}輸出我們指定線程名稱的線程：
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-1 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-10 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-2 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-9 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-5 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-6 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-3 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-7 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-8 Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-4
三、題外話，動(dòng)態(tài)線程池
3.1 什么是動(dòng)態(tài)線程池？
在我們使用線程池的時(shí)候，是否有的時(shí)候很糾結(jié)，到底設(shè)置多大的線程池參數(shù)是最合適的呢？如果不夠用了怎么辦，要改代碼重新部署嗎？
其實(shí)是不需要的，記得當(dāng)初看過(guò)美團(tuán)的一篇文章，真的讓人茅塞頓開(kāi)啊，動(dòng)態(tài)線程池。
ThreadPoolExecutor這個(gè)類其實(shí)是提供對(duì)于線程池的屬性進(jìn)行修改的，支持我們動(dòng)態(tài)修改以下的屬性：
從上至下分別是：
- 線程工廠（用于指定線程名稱）
- 核心線程數(shù)
- 最大線程數(shù)
- 活躍時(shí)間
- 拒絕策略。
在美團(tuán)的文章當(dāng)中呢，是監(jiān)控服務(wù)器線程的使用率，當(dāng)達(dá)到閾值就進(jìn)行告警，然后通過(guò)配置中心去動(dòng)態(tài)修改這些數(shù)值。
我們也可以這么做，使用@RefreshScope加nacos就可以實(shí)現(xiàn)了。
3.2 實(shí)踐
我寫(xiě)了一個(gè)定時(shí)任務(wù)，監(jiān)控當(dāng)前服務(wù)的線程使用率，小了就擴(kuò)容，一段時(shí)間后占用率下降，就恢復(fù)初始值。
其實(shí)還有很多地方需要改進(jìn)的，請(qǐng)大家多提意見(jiàn)，監(jiān)控的是文章前面的線程池GlobalThreadPool,下面調(diào)度任務(wù)的代碼：
/**
* @description： 全局線程池守護(hù)進(jìn)程
* @author：weirx
* @date：2021/9/10 16:32
* @version：3.0
*/
@Slf4j
@Component
public class DaemonThreadTask {
/**
* 服務(wù)支持最大線程數(shù)
*/
public final static int SERVER_MAX_SIZE = 50;
/**
* 最大閾值Maximum threshold，百分比
*/
private final static int MAXIMUM_THRESHOLD = 8;
/**
* 每次遞增最大線程數(shù)
*/
private final static int INCREMENTAL_MAX_NUM = 10;
/**
* 每次遞增核心線程數(shù)
*/
private final static int INCREMENTAL_CORE_NUM = 5;
/**
* 當(dāng)前線程數(shù)
*/
private static int currentSize = GlobalThreadPool.MAX_NUM_POOL_SIZE;
/**
* 當(dāng)前核心線程數(shù)
*/
private static int currentCoreSize = GlobalThreadPool.CORE_POOL_SIZE;
@Scheduled(cron = "0 */5 * * * ?")
public static void execute() {
threadMonitor();
}
/**
* description: 動(dòng)態(tài)監(jiān)控并設(shè)置線程參數(shù)
*
* @return: void
* @author: weirx
* @time: 2021/9/10 13:20
*/
private static void threadMonitor() {
ThreadPoolExecutor instance = GlobalThreadPool.getExecutor();
int activeCount = instance.getActiveCount();
int size = instance.getQueue().size();
log.info("GlobalThreadPool: the active thread count is {}", activeCount);
// 線程數(shù)不足，增加線程
if (activeCount > GlobalThreadPool.MAX_NUM_POOL_SIZE % MAXIMUM_THRESHOLD
&& size >= GlobalThreadPool.BLOCKING_QUEUE_SIZE) {
currentSize = currentSize + INCREMENTAL_MAX_NUM;
currentCoreSize = currentCoreSize + INCREMENTAL_CORE_NUM;
//當(dāng)前設(shè)置最大線程數(shù)小于服務(wù)最大支持線程數(shù)才可以繼續(xù)增加線程
if (currentSize <= SERVER_MAX_SIZE) {
instance.setMaximumPoolSize(currentSize);
instance.setCorePoolSize(currentCoreSize);
log.info("this max thread size is {}", currentSize);
} else {
log.info("current size is more than server max size, can not add");
}
}
// 線程數(shù)足夠，降低線程數(shù)，當(dāng)前活躍數(shù)小于默認(rèn)核心線程數(shù)
if (activeCount < GlobalThreadPool.MAX_NUM_POOL_SIZE
&& size == 0
&& currentSize > GlobalThreadPool.MAX_NUM_POOL_SIZE) {
currentSize = GlobalThreadPool.MAX_NUM_POOL_SIZE;
currentCoreSize = GlobalThreadPool.CORE_POOL_SIZE;
instance.setMaximumPoolSize(currentSize);
instance.setCorePoolSize(currentCoreSize);
}
}
}3.3 動(dòng)態(tài)線程池有什么意義？
有的朋友其實(shí)問(wèn)過(guò)我，我直接把線程池設(shè)置大一點(diǎn)不就好了，這種動(dòng)態(tài)線程池有什么意義呢？
其實(shí)這是一個(gè)好問(wèn)題。在以前的傳統(tǒng)軟件當(dāng)中，單機(jī)部署，硬件部署，確實(shí)，我們能使用的線程數(shù)取決于服務(wù)器的核心線程數(shù)，而且基本沒(méi)有其他服務(wù)來(lái)爭(zhēng)搶這些線程。
但是現(xiàn)在是容器的時(shí)代，云原生的時(shí)代。
多個(gè)容器部署在一個(gè)宿主機(jī)上，那么當(dāng)高峰期的時(shí)候，某個(gè)容器就需要占用大量的cpu資源，如果所有的容器都將大部分資源占據(jù)，那么這個(gè)容器必然面臨阻塞甚至癱瘓的風(fēng)險(xiǎn)。
當(dāng)高峰期過(guò)了，釋放這部分資源可以被釋放掉，用于其他需要的容器。。
再結(jié)合到目前的云服務(wù)器節(jié)點(diǎn)擴(kuò)容，都是需要?jiǎng)討B(tài)擴(kuò)縮容的的，和線程相似，在滿足高可用的情況下，盡量的節(jié)約成本。
總結(jié)
到此這篇關(guān)于java8異步調(diào)用如何使用的文章就介紹到這了,更多相關(guān)java8異步調(diào)用使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
使用注解開(kāi)發(fā)SpringMVC詳細(xì)配置教程
這篇文章主要介紹了使用注解開(kāi)發(fā)SpringMVC詳細(xì)配置教程,本文通過(guò)圖文并茂的形式給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-09-09
這篇文章主要介紹了Mybatis-Plus select不列出全部字段的方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
Java語(yǔ)言實(shí)現(xiàn)簡(jiǎn)單FTP軟件 FTP上傳下載管理模塊實(shí)現(xiàn)(11)
這篇文章主要為大家詳細(xì)介紹了Java語(yǔ)言實(shí)現(xiàn)簡(jiǎn)單FTP軟件，F(xiàn)TP本地文件管理模塊的實(shí)現(xiàn)方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-04-04
java實(shí)現(xiàn)超市商品庫(kù)存管理平臺(tái)
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)超市商品庫(kù)存管理平臺(tái)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-10-10
Java中優(yōu)先隊(duì)列PriorityQueue常用方法示例
這篇文章主要介紹了Java中優(yōu)先隊(duì)列PriorityQueue常用方法示例,PriorityQueue是一種特殊的隊(duì)列,滿足隊(duì)列的“隊(duì)尾進(jìn)、隊(duì)頭出”條件,但是每次插入或刪除元素后,都對(duì)隊(duì)列進(jìn)行調(diào)整,使得隊(duì)列始終構(gòu)成最小堆(或最大堆),需要的朋友可以參考下2023-09-09
Java中的線程到底有哪些安全策略呢？這篇文章將詳細(xì)為大家分析一下。文中的示例代碼講解詳細(xì)，對(duì)我們學(xué)習(xí)或工作有一定幫助，需要的可以參考一下2022-05-05
這篇文章主要為大家介紹了javaWeb傳收參數(shù)方式總結(jié)示例分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-08-08
Springboot項(xiàng)目參數(shù)校驗(yàn)方式(Validator)
本文介紹了如何在Spring Boot項(xiàng)目中使用`spring-boot-starter-validation`包和注解來(lái)實(shí)現(xiàn)請(qǐng)求參數(shù)校驗(yàn),主要介紹了校驗(yàn)注解的使用方法、校驗(yàn)失敗的異常捕獲以及`@Validated`的分組功能2025-02-02
java Struts2框架下實(shí)現(xiàn)文件上傳功能
這篇文章主要為大家詳細(xì)介紹了java Struts2框架下實(shí)現(xiàn)文件上傳功能,具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-10-10
這篇文章主要給大家介紹了關(guān)于java枚舉原來(lái)還可以這么用的相關(guān)資料，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家學(xué)習(xí)或者使用java具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09