1 線程池簡介

1.1 為什么使用線程池

• 降低系統(tǒng)資源消耗，通過重用已存在的線程，降低線程創(chuàng)建和銷毀造成的消耗；
• 提高系統(tǒng)響應(yīng)速度，當(dāng)有任務(wù)到達時，通過復(fù)用已存在的線程，無需等待新線程的創(chuàng)建便能立即執(zhí)行；
• 方便線程并發(fā)數(shù)的管控，因為線程若是無限制的創(chuàng)建，可能會導(dǎo)致內(nèi)存占用過多而產(chǎn)生OOM，并且會造成cpu過度切換（cpu切換線程是有時間成本的（需要保持當(dāng)前執(zhí)行線程的現(xiàn)場，并恢復(fù)要執(zhí)行線程的現(xiàn)場）
• 提供更強大的功能，延時定時線程池

1.2 線程池為什么需要使用隊列

• 因為線程若是無限制的創(chuàng)建，可能會導(dǎo)致內(nèi)存占用過多而產(chǎn)生OOM，并且會造成cpu過度切換。
• 創(chuàng)建線程池的消耗較高或者線程池創(chuàng)建線程需要獲取mainlock這個全局鎖，影響并發(fā)效率，阻塞隊列可以很好的緩沖

1.3 線程池為什么要使用阻塞隊列而不使用非阻塞隊列

• 阻塞隊列可以保證任務(wù)隊列中沒有任務(wù)時阻塞獲取任務(wù)的線程，使得線程進入wait狀態(tài)，釋放cpu資源，當(dāng)隊列中有任務(wù)時才喚醒對應(yīng)線程從隊列中取出消息進行執(zhí)行。
• 使得在線程不至于一直占用cpu資源。（線程執(zhí)行完任務(wù)后通過循環(huán)再次從任務(wù)隊列中取出任務(wù)進行執(zhí)行，代碼片段如：while (task != null || (task = getTask()) != null) {}）。
• 不用阻塞隊列也是可以的，不過實現(xiàn)起來比較麻煩而已，有好用的為啥不用呢

1.4 如何配置線程池

CPU密集型任務(wù)

• 盡量使用較小的線程池，一般為CPU核心數(shù)+1。
• 因為CPU密集型任務(wù)使得CPU使用率很高，若開過多的線程數(shù)，會造成CPU過度切換

IO密集型任務(wù)

• 可以使用稍大的線程池，一般為2*CPU核心數(shù)。
• IO密集型任務(wù)CPU使用率并不高，因此可以讓CPU在等待IO的時候有其他線程去處理別的任務(wù)，充分利用CPU時間

混合型任務(wù)

• 可以將任務(wù)分成IO密集型和CPU密集型任務(wù)，然后分別用不同的線程池去處理。 只要分完之后兩個任務(wù)的執(zhí)行時間相差不大，那么就會比串行執(zhí)行來的高效
• 因為如果劃分之后兩個任務(wù)執(zhí)行時間有數(shù)據(jù)級的差距，那么拆分沒有意義。
• 因為先執(zhí)行完的任務(wù)就要等后執(zhí)行完的任務(wù)，最終的時間仍然取決于后執(zhí)行完的任務(wù)，而且還要加上任務(wù)拆分與合并的開銷，得不償失

1.5 execute()和submit()方法

1.execute()，執(zhí)行一個任務(wù)，沒有返回值

2.submit()，提交一個線程任務(wù)，有返回值

• submit(Callable<T> task)能獲取到它的返回值，通過future.get()獲?。ㄗ枞钡饺蝿?wù)執(zhí)行完）。一般使用FutureTask+Callable配合使用
• submit(Runnable task, T result)能通過傳入的載體result間接獲得線程的返回值。
• submit(Runnable task)則是沒有返回值的，就算獲取它的返回值也是null
• Future.get()方法會使取結(jié)果的線程進入阻塞狀態(tài)，直到線程執(zhí)行完成之后，喚醒取結(jié)果的線程，然后返回結(jié)果

1.6 Spring線程池

Spring 通過任務(wù)執(zhí)行器（TaskExecutor）來實現(xiàn)多線程和并發(fā)編程，使用ThreadPoolTaskExecutor實現(xiàn)一個基于線程池的TaskExecutor，

還得需要使用@EnableAsync開啟異步，并通過在需要的異步方法那里使用注解@Async聲明是一個異步任務(wù)

Spring 已經(jīng)實現(xiàn)的異常線程池：

• SimpleAsyncTaskExecutor：不是真的線程池，這個類不重用線程，每次調(diào)用都會創(chuàng)建一個新的線程。
• SyncTaskExecutor：這個類沒有實現(xiàn)異步調(diào)用，只是一個同步操作。只適用于不需要多線程的地方
• ConcurrentTaskExecutor：Executor的適配類，不推薦使用。如果ThreadPoolTaskExecutor不滿足要求時，才用考慮使用這個類
• SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的類。線程池同時被quartz和非quartz使用，才需要使用此類
• ThreadPoolTaskExecutor ：最常使用，推薦。 其實質(zhì)是對java.util.concurrent.ThreadPoolExecutor的包裝

1.7 @Async調(diào)用中的事務(wù)處理機制

點擊了解使用@Async使用的事務(wù)問題

2 示例

2.1 線程池配置類

package cn.jzh.thread;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
@ComponentScan("cn.jzh.thread")
@EnableAsync ?//開啟異步操作
public class TaskExecutorConfig implements AsyncConfigurer {
? ? /**
? ? ?* 通過getAsyncExecutor方法配置ThreadPoolTaskExecutor,獲得一個基于線程池TaskExecutor
? ? ?*
? ? ?* @return
? ? ?*/
? ? @Override
? ? public Executor getAsyncExecutor() {
? ? ? ? ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
? ? ? ? pool.setCorePoolSize(5);//核心線程數(shù)
? ? ? ? pool.setMaxPoolSize(10);//最大線程數(shù)
? ? ? ? pool.setQueueCapacity(25);//線程隊列
? ? ? ? pool.initialize();//線程初始化
? ? ? ? return pool;
? ? }
? ? @Override
? ? public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
? ? ? ? return null;
? ? }
}

配置類中方法說明：

Spring 中的ThreadPoolExecutor是借助JDK并發(fā)包中的java.util.concurrent.ThreadPoolExecutor來實現(xiàn)的。

其中一些值的含義如下：

• int corePoolSize:線程池維護線程的最小數(shù)量
• int maximumPoolSize:線程池維護線程的最大數(shù)量，線程池中允許的最大線程數(shù)，線程池中的當(dāng)前線程數(shù)目不會超過該值。如果隊列中任務(wù)已滿，并且當(dāng)前線程個數(shù)小于maximumPoolSize，那么會創(chuàng)建新的線程來執(zhí)行任務(wù)。
• long keepAliveTime:空閑線程的存活時間TimeUnit
• unit:時間單位，現(xiàn)由納秒，微秒，毫秒，秒
• BlockingQueue workQueue:持有等待執(zhí)行的任務(wù)隊列，一個阻塞隊列，用來存儲等待執(zhí)行的任務(wù)，當(dāng)線程池中的線程數(shù)超過它的corePoolSize的時候，線程會進入阻塞隊列進行阻塞等待
• RejectedExecutionHandler handler 線程池的拒絕策略，是指當(dāng)任務(wù)添加到線程池中被拒絕，而采取的處理措施。

當(dāng)任務(wù)添加到線程池中之所以被拒絕，可能是由于：第一，線程池異常關(guān)閉。第二，任務(wù)數(shù)量超過線程池的最大限制。

Reject策略預(yù)定義有四種：

• ThreadPoolExecutor.AbortPolicy策略，是默認的策略,處理程序遭到拒絕將拋出運行時 RejectedExecutionException
• ThreadPoolExecutor.CallerRunsPolicy策略 ,調(diào)用者的線程會執(zhí)行該任務(wù),如果執(zhí)行器已關(guān)閉,則丟棄.
• ThreadPoolExecutor.DiscardPolicy策略，不能執(zhí)行的任務(wù)將被丟棄.
• ThreadPoolExecutor.DiscardOldestPolicy策略，如果執(zhí)行程序尚未關(guān)閉，則位于工作隊列頭部的任務(wù)將被刪除，然后重試執(zhí)行程序（如果再次失敗，則重復(fù)此過程）

自定義策略：當(dāng)然也可以根據(jù)應(yīng)用場景需要來實現(xiàn)RejectedExecutionHandler接口自定義策略。如記錄日志或持久化不能處理的任務(wù)

2.2 異步方法

@Async注解可以用在方法上，表示該方法是個異步方法，也可以用在類上，那么表示此類的所有方法都是異步方法

異步方法會自動注入使用ThreadPoolTaskExecutor作為TaskExecutor

package cn.jzh.thread;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import java.util.concurrent.Future;
@Service
public class AsyncTaskService {
? ? /**
? ? ?*?
? ? ?* @param i
? ? ?*/
? ? @Async
? ? public void executeAsync(Integer i) throws Exception{
? ? ? ? System.out.println("線程ID：" + Thread.currentThread().getId() + "線程名字：" +Thread.currentThread().getName()+"執(zhí)行異步任務(wù):" + i);
? ? }
? ? @Async
? ? public Future<String> executeAsyncPlus(Integer i) throws Exception {
? ? ? ? System.out.println("線程ID：" + Thread.currentThread().getId() +"線程名字：" +Thread.currentThread().getName()+ "執(zhí)行異步有返回的任務(wù):" + i);
? ? ? ? return new AsyncResult<>("success:"+i);
? ? }
}

2.3 啟動測試

package cn.jzh.thread;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import java.util.concurrent.Future;
public class MainApp {
? ? public static void main(String[] args) throws Exception{
? ? ? ? System.out.println("主線程id：" + Thread.currentThread().getId() + "開始執(zhí)行調(diào)用任務(wù)...");
? ? ? ? AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(TaskExecutorConfig.class);
? ? ? ? AsyncTaskService service = context.getBean(AsyncTaskService.class);
? ? ? ? for (int i = 0;i<10;i++){
? ? ? ? ? ? service.executeAsync(i);
? ? ? ? ? ? Future<String> result = service.executeAsyncPlus(i);
? ? ? ? ? ? System.out.println("異步程序執(zhí)行結(jié)束，獲取子線程返回內(nèi)容(會阻塞當(dāng)前main線程)" + result.get());
? ? ? ? }
? ? ? ? context.close();
? ? ? ? System.out.println("主線程id：" + Thread.currentThread().getId() + "程序結(jié)束!!");
? ? }
}

注意：

1.是否影響主線程

如果main主線程不去獲取子線程的結(jié)果(Future.get())，那么主線程完全可以不阻塞。那么，此時，主線程和子線程完全異步。此功能，可以做成類似MQ消息中間件之類的，消息異步進行發(fā)送

2.判斷是否執(zhí)行完畢

當(dāng)返回的數(shù)據(jù)類型為Future類型，其為一個接口。具體的結(jié)果類型為AsyncResult,這個是需要注意的地方。

調(diào)用返回結(jié)果的異步方法，判斷是否執(zhí)行完畢時需要使用future.isDone()來判斷是否執(zhí)行完畢

public void testAsyncAnnotationForMethodsWithReturnType() ?
? ?throws InterruptedException, ExecutionException { ?
? ? System.out.println("Invoking an asynchronous method. " ? + Thread.currentThread().getName()); ?
? ? Future<String> future = asyncAnnotationExample.asyncMethodWithReturnType(); ?
? ? while (true) { ?///這里使用了循環(huán)判斷，等待獲取結(jié)果信息 ?
? ? ? ? if (future.isDone()) { ?//判斷是否執(zhí)行完畢 ?
? ? ? ? ? ? System.out.println("Result from asynchronous process - " + future.get()); ?
? ? ? ? ? ? break; ?
? ? ? ? } ?
? ? ? ? System.out.println("Continue doing something else. "); ?
? ? ? ? Thread.sleep(1000); ?
? ? } ?
}

這些獲取異步方法的結(jié)果信息，是通過不停的檢查Future的狀態(tài)來獲取當(dāng)前的異步方法是否執(zhí)行完畢來實現(xiàn)的

總結(jié)

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論