1 前言

說(shuō)到線(xiàn)程池八股文背的很熟的肯定知道無(wú)非就這幾個(gè)考點(diǎn)：

（1）線(xiàn)程池三大核心參數(shù) corePoolSize、maximumPoolSize、workQueue 的含義

（2）線(xiàn)程池核心線(xiàn)程數(shù)制定策略

（3）建議通過(guò) ThreadPoolExecutor 的構(gòu)造函數(shù)來(lái)聲明，避免使用 Executors 創(chuàng)建線(xiàn)程池

以上考點(diǎn)作為線(xiàn)程池面試幾乎必問(wèn)的內(nèi)容，大部分人應(yīng)該都是如數(shù)家珍，張口就來(lái)，但是懂了面試八股文真的就不一定在實(shí)際運(yùn)用中真的就會(huì)把線(xiàn)程池用好 。且看下面這次真實(shí)生產(chǎn)事故還原

2 事故還原

某次一位研發(fā)同事寫(xiě)出了下面類(lèi)似的代碼：

Listitems=getFromDb();
List>completableFutures=items.stream().map(item->CompletableFuture.supplyAsync(()->{
AppMapStationDatadata=mapper.copy(item);
//發(fā)起價(jià)格信息查詢(xún)的RPC調(diào)用
data.setPriceInfo(priceApi.getPriceInfoById(item.getId()))
returndata;
},apiExecutor)).collect(Collectors.toList());

result=completableFutures.stream().map(e->{
returne.get();
}).filter(Objects::nonNull).collect(Collectors.toList());

上面的代碼中，代碼首先從數(shù)據(jù)庫(kù)里面查出來(lái)一堆對(duì)象，然后對(duì)每一個(gè)對(duì)象進(jìn)行模型轉(zhuǎn)換，由于要獲取每個(gè)對(duì)象的價(jià)格信息發(fā)起了一次RPC調(diào)用，由于RPC服務(wù)沒(méi)有提供批量接口，所以代碼里面用了線(xiàn)程池并發(fā)請(qǐng)求，以求得接口盡可能快的返回?cái)?shù)據(jù)。

使用的是CompletableFuture 而且自定義了線(xiàn)程池，線(xiàn)程池指定了10個(gè)核心線(xiàn)程，20個(gè)最大線(xiàn)程，這段代碼在上線(xiàn)后的一段時(shí)間確實(shí)沒(méi)有任何問(wèn)題，但是在灰度放量用戶(hù)量多起來(lái)之后發(fā)現(xiàn)接口經(jīng)常超時(shí)告警。

請(qǐng)問(wèn)為什么上面的代碼在用戶(hù)量稍微大一點(diǎn)的時(shí)候就運(yùn)行緩慢了呢？

實(shí)際代碼問(wèn)題出現(xiàn)在了這個(gè)get方法中，這個(gè)get方法沒(méi)有指定超時(shí)時(shí)間，當(dāng)getPriceInfoById這個(gè)接口響應(yīng)變慢的時(shí)候，這個(gè)主線(xiàn)程的代碼get又沒(méi)有指定超時(shí)時(shí)間，這時(shí)候問(wèn)題就來(lái)了。

由于某次業(yè)務(wù)查詢(xún)查到了非常多的數(shù)據(jù)，每條數(shù)據(jù)就是個(gè)模型轉(zhuǎn)換任務(wù)，這個(gè)任務(wù)就會(huì)在隊(duì)列排隊(duì)，get方法沒(méi)有指定超時(shí)時(shí)間的情況下，其最終耗時(shí)就取決于整個(gè)線(xiàn)程池中執(zhí)行最慢的那一個(gè)任務(wù)，所以當(dāng)從DB中查出來(lái)的數(shù)據(jù)量越來(lái)越大的時(shí)候這個(gè)轉(zhuǎn)換任務(wù)的最大耗時(shí)就會(huì)逐漸增加，進(jìn)而引發(fā)接口超時(shí)。

所以這里改進(jìn)上述問(wèn)題需要做到兩個(gè)點(diǎn)：

1、數(shù)據(jù)庫(kù)中查出來(lái)的數(shù)據(jù)集合必須分頁(yè)

2、get方法必須設(shè)置超時(shí)時(shí)間

此外需要知道get方法設(shè)置超時(shí)時(shí)間的計(jì)算方式也需要留意，考慮下面這種場(chǎng)景

提交兩個(gè)任務(wù) A 和 B 到線(xiàn)程池，A 任務(wù)耗時(shí) 3 秒，B 任務(wù)耗時(shí) 4 秒，F(xiàn)uture 以 2 秒為超時(shí)時(shí)間獲取任務(wù)結(jié)果

代碼如下：

ExecutorServiceexecutorService=Executors.newFixedThreadPool(2);

CallabletaskA=()->{
sleep(3);
return"A";
};
CallabletaskB=()->{
sleep(4);
return"B";
};

List>futures=Stream.of(taskA,taskB)
.map(executorService::submit)
.collect(Collectors.toList());

for(Futurefuture:futures){
try{
Strings=future.get(2,TimeUnit.SECONDS);
System.out.println(s);
}catch(Exceptione){
continue;
}
}

實(shí)際運(yùn)行情況是第一個(gè)任務(wù)會(huì)超時(shí)但是第二個(gè)不會(huì) ，看起來(lái)是不是還有點(diǎn)不可思議，耗時(shí)時(shí)間長(zhǎng)的任務(wù)B反而沒(méi)超時(shí)。原因就在于 Future.get(long timeout, TimeUnit unit) ，調(diào)用 get 時(shí)才開(kāi)始計(jì)時(shí)，而非任務(wù)加入線(xiàn)程池的時(shí)間

從圖上就可以看出來(lái)，在獲取B的任務(wù)執(zhí)行結(jié)果的時(shí)候B任務(wù)已經(jīng)執(zhí)行了兩秒，所以在等待兩秒的情況下可以獲取到結(jié)果

3 線(xiàn)程池不當(dāng)使用舉例

（1）不區(qū)分業(yè)務(wù)一把梭哈，全用一個(gè)線(xiàn)程池

曾經(jīng)有一個(gè)項(xiàng)目，對(duì)接多個(gè)租戶(hù)，每個(gè)租戶(hù)都有各自的任務(wù)需要執(zhí)行，代碼中不區(qū)分租戶(hù)的將所有租戶(hù)的任務(wù)全部丟到一個(gè)線(xiàn)程池中執(zhí)行，結(jié)果一個(gè)租戶(hù)的任務(wù)提交過(guò)多導(dǎo)致線(xiàn)程池執(zhí)行緩慢，但是由于線(xiàn)程池是同一個(gè)，影響了所有租戶(hù)接口的響應(yīng)時(shí)間。如果說(shuō)上面說(shuō)的這個(gè)場(chǎng)景用一個(gè)線(xiàn)程池產(chǎn)生了租戶(hù)互相影響的問(wèn)題還不夠嚴(yán)重，那么下面的這種場(chǎng)景就問(wèn)題更大了。

曾經(jīng)有一段這樣的場(chǎng)景，因?yàn)楣灿镁€(xiàn)程池直接導(dǎo)致線(xiàn)程池任務(wù)永遠(yuǎn)完成不了，請(qǐng)看下面的這種情況：

首先向線(xiàn)程池中提交了一個(gè)任務(wù)，然后在這個(gè)任務(wù)的內(nèi)部實(shí)現(xiàn)中又往同一個(gè)線(xiàn)程池中再次提交了一個(gè)任務(wù)，相當(dāng)于父子任務(wù)在同一個(gè)線(xiàn)程池中執(zhí)行，這時(shí)候極有可出現(xiàn)線(xiàn)程死鎖也就是循環(huán)等待的情況

如上圖所示，父任務(wù)全部處于執(zhí)行狀態(tài)，這時(shí)候子任務(wù)想要執(zhí)行需要等父任務(wù)執(zhí)行完成，但是父任務(wù)都執(zhí)行不完，因?yàn)檫€有個(gè)子任務(wù)沒(méi)完成，即父任務(wù)等待子任務(wù)執(zhí)行完成，而子任務(wù)等待父任務(wù)釋放線(xiàn)程池資源，這也就造成了 "死鎖"

所以綜上所述，在代碼中應(yīng)該避免各種任務(wù)都往一個(gè)線(xiàn)程池中投放，對(duì)每個(gè)線(xiàn)程池指定好線(xiàn)程名稱(chēng)，做好分類(lèi)比較合適，這里在日常開(kāi)發(fā)中比較推薦使用Guava的工具類(lèi)，來(lái)指定線(xiàn)程名稱(chēng)前綴，這樣使用jstack分析線(xiàn)程問(wèn)題也方便排查。

ThreadFactorythreadFactory=newThreadFactoryBuilder()
.setNameFormat(threadNamePrefix+"-%d")
.setDaemon(true).build();
ExecutorServicethreadPool=newThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
TimeUnit.MINUTES,
workQueue,
threadFactory);

（2）@Async注解不自己定義線(xiàn)程池

@Async用在方法上標(biāo)識(shí)這是一個(gè)異步方法，如果不自己指定線(xiàn)程池這個(gè)方法將直接新建一個(gè)線(xiàn)程執(zhí)行，可以翻看spring實(shí)現(xiàn)源碼知道這個(gè)點(diǎn)

@Async的實(shí)現(xiàn)其實(shí)非常簡(jiǎn)單就是利用AOP，容器啟動(dòng)的時(shí)候會(huì)掃描所有被打上@Async注解的方法，并代理這些方法的執(zhí)行，在執(zhí)行這個(gè)方法的時(shí)候，生成Callable任務(wù)丟到線(xiàn)程池中執(zhí)行（核心代碼位于org.springframework.aop.interceptor.AsyncExecutionInterceptor）

@Override
@Nullable
publicObjectinvoke(finalMethodInvocationinvocation)throwsThrowable{
ClasstargetClass=(invocation.getThis()!=null?AopUtils.getTargetClass(invocation.getThis()):null);
MethodspecificMethod=ClassUtils.getMostSpecificMethod(invocation.getMethod(),targetClass);
finalMethoduserDeclaredMethod=BridgeMethodResolver.findBridgedMethod(specificMethod);

AsyncTaskExecutorexecutor=determineAsyncExecutor(userDeclaredMethod);
if(executor==null){
thrownewIllegalStateException(
"NoexecutorspecifiedandnodefaultexecutorsetonAsyncExecutionInterceptoreither");
}
//將方法調(diào)用封裝成Callable實(shí)例丟入線(xiàn)程池中執(zhí)行
Callable