簡(jiǎn)介

在java中，除了單個(gè)使用Thread之外，我們還會(huì)使用到ThreadPool來(lái)構(gòu)建線程池，那么在使用線程池的過(guò)程中需要注意哪些事情呢？

一起來(lái)看看吧。

java自帶的線程池

java提供了一個(gè)非常好用的工具類Executors，通過(guò)Executors我們可以非常方便的創(chuàng)建出一系列的線程池：

Executors.newCachedThreadPool,根據(jù)需要可以創(chuàng)建新線程的線程池。線程池中曾經(jīng)創(chuàng)建的線程，在完成某個(gè)任務(wù)后也許會(huì)被用來(lái)完成另外一項(xiàng)任務(wù)。

Executors.newFixedThreadPool(int nThreads) ,創(chuàng)建一個(gè)可重用固定線程數(shù)的線程池。這個(gè)線程池里最多包含nThread個(gè)線程。

Executors.newSingleThreadExecutor() ,創(chuàng)建一個(gè)使用單個(gè) worker 線程的 Executor。即使任務(wù)再多，也只用1個(gè)線程完成任務(wù)。

Executors.newSingleThreadScheduledExecutor() ,創(chuàng)建一個(gè)單線程執(zhí)行程序，它可安排在給定延遲后運(yùn)行命令或者定期執(zhí)行。

提交給線程池的線程要是可以被中斷的

ExecutorService線程池提供了兩個(gè)很方便的停止線程池中線程的方法，他們是shutdown和shutdownNow。

shutdown不會(huì)接受新的任務(wù)，但是會(huì)等待現(xiàn)有任務(wù)執(zhí)行完畢。而shutdownNow會(huì)嘗試立馬終止現(xiàn)有運(yùn)行的線程。

那么它是怎么實(shí)現(xiàn)的呢？我們看一個(gè)ThreadPoolExecutor中的一個(gè)實(shí)現(xiàn)：

public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP); interruptWorkers(); tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; }

里面有一個(gè)interruptWorkers()方法的調(diào)用，實(shí)際上就是去中斷當(dāng)前運(yùn)行的線程。

所以我們可以得到一個(gè)結(jié)論，提交到ExecutorService中的任務(wù)一定要是可以被中斷的，否則shutdownNow方法將會(huì)失效。

先看一個(gè)錯(cuò)誤的使用例子：

public void wrongSubmit(){ Runnable runnable= ()->{ try(SocketChannel sc = SocketChannel.open(new InetSocketAddress('127.0.0.1', 8080))) { ByteBuffer buf = ByteBuffer.allocate(1024); while(true){sc.read(buf); } } catch (IOException e) {e.printStackTrace(); } }; ExecutorService pool = Executors.newFixedThreadPool(10); pool.submit(runnable); pool.shutdownNow(); }

在這個(gè)例子中，運(yùn)行的代碼無(wú)法處理中斷，所以將會(huì)一直運(yùn)行。

下面看下正確的寫法：

public void correctSubmit(){ Runnable runnable= ()->{ try(SocketChannel sc = SocketChannel.open(new InetSocketAddress('127.0.0.1', 8080))) {ByteBuffer buf = ByteBuffer.allocate(1024);while(!Thread.interrupted()){ sc.read(buf);} } catch (IOException e) {e.printStackTrace(); } }; ExecutorService pool = Executors.newFixedThreadPool(10); pool.submit(runnable); pool.shutdownNow(); }

我們需要在while循環(huán)中加上中斷的判斷，從而控制程序的執(zhí)行。

正確處理線程池中線程的異常

如果在線程池中的線程發(fā)生了異常，比如RuntimeException，我們?cè)趺床拍軌虿蹲降侥兀?如果不能夠?qū)Ξ惓＿M(jìn)行合理的處理，那么將會(huì)產(chǎn)生不可預(yù)料的問(wèn)題。

看下面的例子：

public void wrongSubmit() throws InterruptedException { ExecutorService pool = Executors.newFixedThreadPool(10); Runnable runnable= ()->{ throw new NullPointerException(); }; pool.execute(runnable); Thread.sleep(5000); System.out.println('finished!'); }

上面的例子中，我們submit了一個(gè)任務(wù)，在任務(wù)中會(huì)拋出一個(gè)NullPointerException，因?yàn)槭欠莄hecked異常，所以不需要顯式捕獲，在任務(wù)運(yùn)行完畢之后，我們基本上是不能夠得知任務(wù)是否運(yùn)行成功了。

那么，怎么才能夠捕獲這樣的線程池異常呢？這里介紹大家?guī)讉€(gè)方法。

第一種方法就是繼承ThreadPoolExecutor，重寫

protected void afterExecute(Runnable r, Throwable t) { }

和

protected void terminated() { }

這兩個(gè)方法。

其中afterExecute會(huì)在任務(wù)執(zhí)行完畢之后被調(diào)用，Throwable t中保存的是可能出現(xiàn)的運(yùn)行時(shí)異常和Error。我們可以根據(jù)需要進(jìn)行處理。

而terminated是在線程池中所有的任務(wù)都被調(diào)用完畢之后才被調(diào)用的。我們可以在其中做一些資源的清理工作。

第二種方法就是使用UncaughtExceptionHandler。

Thread類中提供了一個(gè)setUncaughtExceptionHandler方法，用來(lái)處理捕獲的異常，我們可以在創(chuàng)建Thread的時(shí)候，為其添加一個(gè)UncaughtExceptionHandler就可以了。

但是ExecutorService執(zhí)行的是一個(gè)個(gè)的Runnable，怎么使用ExecutorService來(lái)提交Thread呢？

別怕， Executors在構(gòu)建線程池的時(shí)候，還可以讓我們傳入ThreadFactory，從而構(gòu)建自定義的Thread。

public void useExceptionHandler() throws InterruptedException { ThreadFactory factory =new ExceptionThreadFactory(new MyExceptionHandler()); ExecutorService pool =Executors.newFixedThreadPool(10, factory); Runnable runnable= ()->{ throw new NullPointerException(); }; pool.execute(runnable); Thread.sleep(5000); System.out.println('finished!'); } public static class ExceptionThreadFactory implements ThreadFactory { private static final ThreadFactory defaultFactory =Executors.defaultThreadFactory(); private final Thread.UncaughtExceptionHandler handler; public ExceptionThreadFactory(Thread.UncaughtExceptionHandler handler) { this.handler = handler; } @Override public Thread newThread(Runnable run) { Thread thread = defaultFactory.newThread(run); thread.setUncaughtExceptionHandler(handler); return thread; } } public static class MyExceptionHandler implements Thread.UncaughtExceptionHandler { @Override public void uncaughtException(Thread t, Throwable e) { } }

上面的例子有點(diǎn)復(fù)雜了， 有沒(méi)有更簡(jiǎn)單點(diǎn)的做法呢？

有的。ExecutorService除了execute來(lái)提交任務(wù)之外，還可以使用submit來(lái)提交任務(wù)。不同之處是submit會(huì)返回一個(gè)Future來(lái)保存執(zhí)行的結(jié)果。

public void useFuture() throws InterruptedException { ExecutorService pool = Executors.newFixedThreadPool(10); Runnable runnable= ()->{ throw new NullPointerException(); }; Future future = pool.submit(runnable); try { future.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } Thread.sleep(5000); System.out.println('finished!'); }

當(dāng)我們?cè)谡{(diào)用future.get()來(lái)獲取結(jié)果的時(shí)候，異常也會(huì)被封裝到ExecutionException，我們可以直接獲取到。

線程池中使用ThreadLocal一定要注意清理

我們知道ThreadLocal是Thread中的本地變量，如果我們?cè)诰€程的運(yùn)行過(guò)程中用到了ThreadLocal，那么當(dāng)線程被回收之后再次執(zhí)行其他的任務(wù)的時(shí)候就會(huì)讀取到之前被設(shè)置的變量，從而產(chǎn)生未知的問(wèn)題。

正確的使用方法就是在線程每次執(zhí)行完任務(wù)之后，都去調(diào)用一下ThreadLocal的remove操作。

或者在自定義ThreadPoolExecutor中，重寫beforeExecute(Thread t, Runnable r)方法，在其中加入ThreadLocal的remove操作。

本文的代碼：

https://github.com/ddean2009/learn-java-base-9-to-20/tree/master/security

以上就是Java ThreadPool的使用解析的詳細(xì)內(nèi)容，更多關(guān)于Java ThreadPool的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！