Java中的線程池

1.簡介

 

使用線程池能夠避免線程的頻繁建立以及銷燬。

JAVA中提供的用於實現線程池的API：

 

 

 

Executor、ExecutorService、AbstractExecutorService、ThreadPoolExecutor、ForkJoinPool都位於java.util.concurrent包下。

*ThreadPoolExecutor、ForkJoinPool爲線程池的實現類。 

 

 

2.Executor

 

public interface Executor { /** * 向線程池提交一個任務,交由線程池去執行 */
    void execute(Runnable command); }

 

*該接口聲明瞭execute(Runnable command)方法，負責向線程池中提交一個任務。

 

 

3.ExecutorService接口

 

public interface ExecutorService extends Executor { /** * 關閉線程池(等待隊列中的任務被執行完畢) */
    void shutdown(); /** * 馬上關閉線程池(不執行隊列中的任務，並嘗試中斷當前執行的任務) */ List<Runnable> shutdownNow(); /** * 判斷線程池是否處於shutdown狀態. */
    boolean isShutdown(); /** * 判斷線程池是否處於terminated狀態. */
    boolean isTerminated(); /** * 若在指定時間內線程池處於terminated狀態則當即返回true,不然超過期間後仍未爲terminated狀態則返回false. */
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; /** * 向線程池提交一個任務並返回包含指定類型的Future(根據Callable的泛型) */
    <T> Future<T> submit(Callable<T> task); /** * 向線程池提交一個任務並指定任務執行結果的類型,返回包含指定類型的Future. */
    <T> Future<T> submit(Runnable task, T result); /** * 向線程池提交一個任務並返回未知類型的Future. */ Future<?> submit(Runnable task); /** * 向線程池提交多個任務並返回指定類型的Future列表. */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; /** * 向線程池提交多個任務並返回指定類型的Future列表,若是在指定時間內沒有執行完畢則直接返回. */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException; /** * 向線程池提交多個任務,當任意一個任務執行完畢後返回指定類型的Future. */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException; /** * 向線程池提交多個任務,在指定時間內,當任意一個任務執行完畢後返回指定類型的Future,若超時則拋出異常. */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

 

public interface Future<V> { /** * 中斷任務的執行 */
    boolean cancel(boolean mayInterruptIfRunning); /** * 判斷任務是否中斷成功 */
    boolean isCancelled(); /** * 判斷任務是否執行完成 */
    boolean isDone(); /** * 獲取任務的執行結果直到任務執行完畢(阻塞線程) */ V get() throws InterruptedException, ExecutionException; /** * 獲取任務的執行結果,若在指定時間內任務仍然沒有執行完畢則拋出TimeoutException */ V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

 

*execute()方法不能獲取任務的執行結果，而submit()方法可以根據返回的Future實例獲取任務的執行結果。

 

 

4.ThreadPoolExecutor

 

 

ThreadPoolExecutor聲明的核心屬性

 

corePoolSize：線程池中核心線程的數量。

maximumPoolSize：線程池中最大線程數。

keepAliveTime：線程的空閒時間。

unit：修飾線程空閒時間的單位。

workQueue：任務隊列。

threadFactory：線程工廠，用於建立線程。

handler：當隊列已滿且當前線程數已達到所容許的最大值時的處理策略。

*線程池中的線程包括核心線程以及普通線程，核心線程一旦建立後直到線程池被關閉前都就不會被銷燬，而普通線程會由於到達空閒時間而被銷燬。

 

構造方法：

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

 

 

BlockingQueue的類型 

 

BlockingQueue提供了ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue等實現類。

1.ArrayBlockingQueue：使用順序表的結構進行存儲，在使用時須要指定其長度，支持公平鎖/非公平鎖進行操做。

2.LinkedBlockingQueue：使用鏈表的結構進行存儲，在使用時不須要指定其長度，隊列的最大長度爲Integer.MAX_VALUE。

3.SynchronousQueue：一個不存儲元素的隊列，每個put操做必須等待take操做，不然不能添加元素，支持公平鎖和非公平鎖。

 

*這些實現類在進行入隊和出隊操做時都會進行加鎖，以保證在多線程併發訪問時數據的安全性。

 

 

隊列已滿且線程數已達到所容許的最大值時的處理策略

 

RejectedExecutionHandler提供了AbortPolicy、DiscardPolicy、DiscardOlderstPolicy、CallerRunsPolicy四個策略，這四個策略都是ThreadPoolExecutor的靜態內部類。

1.AbortPolicy：放棄任務並拋出RejectedExecutionException異常。

2.DiscardPolicy：放棄任務但不拋出異常。

3.DiscardOlderstPolicy： 放棄隊頭中的任務，而後從新嘗試執行新任務。

4.CallerRunsPolicy： 由調用線程來處理該任務。

 

 

線程池的狀態

 

private static final int RUNNING    = -1；
private static final int SHUTDOWN   =  0； 
private static final int STOP       =  1；
private static final int TIDYING    =  2； 
private static final int TERMINATED =  3；

 

1.RUNING：線程池處於運行狀態，此時能夠接受新的任務請求，而且執行隊列中的任務。

2.SHUTDOWN：線程池處於關閉狀態，此時不接受新的任務請求，但會繼續執行隊列中的任務。

3.STOP：線程池處於禁用狀態，此時不接受新的任務請求，而且不會執行隊列中的任務。

4.TIDYING：線程池處於整理狀態，此時沒有正在執行的任務。

5.TERMINATED ：線程池處於終止狀態。

 

線程池狀態的變化過程

1.當線程池建立後處於RUNNING狀態。

2.1 若此時調用了shutdown()方法，那麼線程池將處於SHUTDOWN狀態，不接受新的任務請求，但會繼續執行隊列中的任務，當隊列中的任務爲空且沒有正在執行的任務時，線程池的狀態爲TIDYING。

2.2 若此時調用了shutdownNow()方法，那麼線程池將處於STOP狀態，不接受新的任務請求而且不執行隊列中的任務，此時線程池的狀態爲TIDYING。

3.當線程池的狀態爲TIDYING時，當terminated()方法處理完畢後，線程池的狀態爲TRRMINATED。

 

 

任務的執行流程

 

1.當調用了execute()或者submit()方法向線程池提交一個任務後，首先判斷當前線程池中的線程個數是否大於核心線程數。

2.若是當前線程池的線程個數小於核心線程數，則建立一個核心線程來處理任務。

3.若是當前線程池的線程個數大於核心線程數，則將任務放入到隊列中，若是放入隊列成功，那麼該任務將等待被空閒的線程處理，若是放入隊列失敗(隊滿)，則判斷當前線程池中的線程個數是否達到所容許的最大值，若未達到則建立一個普通線程去處理任務，不然根據預約義的處理策略去進行處理。

 

 

 

 

 

5.Executors工具類

 

JAVA中提供了Executors工具類，用於直接建立Executor。

 

CacheThreadPool

public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); }

CacheThreadPool建立的都是普通線程(其核心線程數爲0)、線程池的最大線程數爲Integer.MAX_VALUE、線程的空閒時間爲60秒，此方式適合大量耗時短的任務、不適合大量耗時長的任務。

*因爲建立的都是普通線程，且空閒時間爲60秒，則仍有可能會頻繁的建立線程。

 

FixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()); }

FixedThreadPool建立的都是核心線程，其線程個數由入參決定，線程不會由於空閒時間而被銷燬，適合預知任務數量的業務。

 

SingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,new LinkedBlockingQueue<Runnable>())); }

SingleThreadExecutor使用一個核心線程來處理任務。

 

ScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); }

*ScheduledThreadPool支持定時執行任務以及固定間隔執行任務。

 

SingleThreadScheduledExecutor

public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1)); }

*SingleThreadScheduledExecutor支持一個線程的定時執行任務以及固定間隔執行任務。

 

public interface ScheduledExecutorService extends ExecutorService { /** * 在指定的延遲時間到達後執行任務一次 */
    public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit); /** * 在指定的延遲時間到達後執行任務一次 */
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit); /** * 在指定的初始化延遲時間到達後執行任務一次,日後每隔period時間執行任務一次. */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit); /** * 在指定的初始化延遲時間到達後執行任務一次,日後每次任務執行完畢後相隔delay時間執行任務一次. */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit); }

 

WorkStealingPool

public static ExecutorService newWorkStealingPool(int parallelism) { return new ForkJoinPool(parallelism,ForkJoinPool.defaultForkJoinWorkerThreadFactory,null, true); }

WorkStealingPool建立一個並行級別的線程池，同一時刻最多隻能有指定個數個線程正在執行任務，建立時直接指定同一時刻最多能容許的並行執行的線程個數便可，若是不傳則使用CPU的核數。

newWorkStealingPool方法內部返回一個ForkJoinPool實例，ForkJoinPool是JAVA7新提供的線程池，一樣繼承AbstactExecutorService。

*做用相似於Semaphore。