ExecutorService與Executors例子的簡單剖析

baopbird2005

對于多線程有了一點了解之后，那么來看看java.lang.concurrent包下面的一些東西。在此之前，我們運行一個線程都是顯式調(diào)用了Thread的start()方法。我們用concurrent下面的類來實現(xiàn)一下線程的運行，而且這將成為以后常用的方法或者實現(xiàn)思路。

        看一個簡單的例子：

1. public class CacheThreadPool {  
   
2.     public static void main(String[] args) {  
   
3.         ExecutorService exec=Executors.newCachedThreadPool();  
   
4.         for(int i=0;i<5;i++)  
   
5.             exec.execute(new LiftOff());  
   
6.         exec.shutdown();//并不是終止線程的運行，而是禁止在這個Executor中添加新的任務  
   
7.     }  
   
8. }

復制代碼

這個例子其實很容易看懂，ExecutorService中有一個execute方法，這個方法的參數(shù)是Runnable類型。也就是說，將一個實現(xiàn)了Runnable類型的類的實例作為參數(shù)傳入execute方法并執(zhí)行，那么線程就相應的執(zhí)行了。

        一、ExecutorService
        先看看ExecutorService，這是一個接口，簡單的列一下這個接口：

1. public interface ExecutorService extends Executor {  
   
2.   
   
3.     void shutdown();  
   
4.   
   
5.     List<Runnable> shutdownNow();  
   
6.   
   
7.     boolean isShutdown();  
   
8.   
   
9.     boolean isTerminated();  
   
10.   
    
11.     boolean awaitTermination(long timeout, TimeUnit unit)  
    
12.   
    
13.     <T> Future<T> submit(Callable<T> task);  
    
14.   
    
15.     <T> Future<T> submit(Runnable task, T result);  
    
16.   
    
17.     Future<?> submit(Runnable task);  
    
18.   
    
19.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)  
    
20.   
    
21.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)  
    
22.   
    
23.     <T> T invokeAny(Collection<? extends Callable<T>> tasks)  
    
24.   
    
25.     <T> T invokeAny(Collection<? extends Callable<T>> tasks,  
    
26.                     long timeout, TimeUnit unit)  
    
27. }

復制代碼

ExecuteService繼承了Executor，Executor也是一個接口，里面只有一個方法：

1. void execute(Runnable command)

復制代碼

二、Executors
        Executors是一個類，直接援引JDK文檔的說明來說一下這個類的作用：
      

        Factory and utility methods for Executor, ExecutorService, ScheduledExecutorService, ThreadFactory, and Callable classes defined in this package. This class supports the following kinds of methods:


      
Methods that create and return an ExecutorService set up with commonly useful configuration settings.
      
Methods that create and return a ScheduledExecutorService set up with commonly useful configuration settings.
      
Methods that create and return a "wrapped" ExecutorService, that disables reconfiguration by making implementation-specific methods inaccessible.
      
Methods that create and return a ThreadFactory that sets newly created threads to a known state.
      
Methods that create and return a Callable out of other closure-like forms, so they can be used in execution methods requiring Callable.



在上面的例子中，我們用到了newCachedThreadPool()方法�？匆幌逻@個方法：

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

復制代碼

在源碼中我們可以知道兩點，1、這個方法返回類型是ExecutorService；2、此方法返回值實際是另一個類的實例�？匆幌逻@個類的信息：

1. public class ThreadPoolExecutor extends AbstractExecutorService {  
   
2.     ..........  
   
3.     private final BlockingQueue<Runnable> workQueue;//這個變量在下面會提到  
   
4.     ..........  
   
5. }  

復制代碼

ThreadPoolExecutor繼承了AbstractExecutorService,而AbstractExecutorService又實現(xiàn)了ExecutorService接口。所以，根據(jù)多態(tài)，ThreadPoolExecutor可以看作是ExecutorService類型。

        線程執(zhí)行的最關鍵的一步是執(zhí)行了executor方法，根據(jù)java的動態(tài)綁定，實際執(zhí)行的是ThreadPoolExecutor所實現(xiàn)的executor方法�？纯丛创a：

1. public class ThreadPoolExecutor extends AbstractExecutorService {  
   
2.     ..........  
   
3.     public void execute(Runnable command) {  
   
4.         if (command == null)  
   
5.             throw new NullPointerException();  
   
6.         if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  
   
7.             if (runState == RUNNING && workQueue.offer(command)) {  
   
8.                 if (runState != RUNNING || poolSize == 0)  
   
9.                     ensureQueuedTaskHandled(command);  
   
10.             }  
    
11.             else if (!addIfUnderMaximumPoolSize(command))  
    
12.                 reject(command); // is shutdown or saturated  
    
13.         }  
    
14.     }  
    
15.     ..........  
    
16. }

復制代碼

根據(jù)程序正常執(zhí)行的路線來看，這個方法中比較重要的兩個地方分別是：
        1、workQueue.offer(command)
        workQueue在上面提到過，是BlockingQueue<Runnable>類型的變量，這條語句就是將Runnable類型的實例加入到隊列中。
        2、ensureQueuedTaskHandled(command)
        這個是線程執(zhí)行的關鍵語句�？纯此�的源碼：

1. public class ThreadPoolExecutor extends AbstractExecutorService {  
   
2.     ..........  
   
3.     private void ensureQueuedTaskHandled(Runnable command) {  
   
4.         final ReentrantLock mainLock = this.mainLock;  
   
5.         mainLock.lock();  
   
6.         boolean reject = false;  
   
7.         Thread t = null;  
   
8.         try {  
   
9.             int state = runState;  
   
10.             if (state != RUNNING && workQueue.remove(command))  
    
11.                 reject = true;  
    
12.             else if (state < STOP &&  
    
13.                      poolSize < Math.max(corePoolSize, 1) &&  
    
14.                      !workQueue.isEmpty())  
    
15.                 t = addThread(null);  
    
16.         } finally {  
    
17.             mainLock.unlock();  
    
18.         }  
    
19.         if (reject)  
    
20.             reject(command);  
    
21.         else if (t != null)  
    
22.             t.start();  
    
23.     }  
    
24.     ..........  
    
25. }  

復制代碼

在這里我們就可以看到最終執(zhí)行了t.start()方法來運行線程。在這之前的重點是t=addThread(null)方法，看看addThread方法的源碼：

1. public class ThreadPoolExecutor extends AbstractExecutorService {  
   
2.     ..........  
   
3.     private Thread addThread(Runnable firstTask) {  
   
4.         Worker w = new Worker(firstTask);  
   
5.         Thread t = threadFactory.newThread(w);  
   
6.         if (t != null) {  
   
7.             w.thread = t;  
   
8.             workers.add(w);  
   
9.             int nt = ++poolSize;  
   
10.             if (nt > largestPoolSize)  
    
11.                 largestPoolSize = nt;  
    
12.         }  
    
13.         return t;  
    
14.     }  
    
15.     ..........  
    
16. }

復制代碼

這里兩個重點，很明顯：
        1、Worker w = new Worker(firstTask)
        2、Thread t = threadFactory.newThread(w)
        先看Worker是個什么結構：

1. public class ThreadPoolExecutor extends AbstractExecutorService {  
   
2.     ..........  
   
3.     private final class Worker implements Runnable {  
   
4.         ..........  
   
5.         Worker(Runnable firstTask) {  
   
6.             this.firstTask = firstTask;  
   
7.         }  
   
8.   
   
9.         private Runnable firstTask;  
   
10.         ..........  
    
11.   
    
12.         public void run() {  
    
13.             try {  
    
14.                 Runnable task = firstTask;  
    
15.                 firstTask = null;  
    
16.                 while (task != null || (task = getTask()) != null) {  
    
17.                     runTask(task);  
    
18.                     task = null;  
    
19.                 }  
    
20.             } finally {  
    
21.                 workerDone(this);  
    
22.             }  
    
23.         }  
    
24.     }  
    
25.   
    
26.     Runnable getTask() {  
    
27.         for (;;) {  
    
28.             try {  
    
29.                 int state = runState;  
    
30.                 if (state > SHUTDOWN)  
    
31.                     return null;  
    
32.                 Runnable r;  
    
33.                 if (state == SHUTDOWN)  // Help drain queue  
    
34.                     r = workQueue.poll();  
    
35.                 else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  
    
36.                     r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);  
    
37.                 else  
    
38.                     r = workQueue.take();  
    
39.                 if (r != null)  
    
40.                     return r;  
    
41.                 if (workerCanExit()) {  
    
42.                     if (runState >= SHUTDOWN) // Wake up others  
    
43.                         interruptIdleWorkers();  
    
44.                     return null;  
    
45.                 }  
    
46.                 // Else retry  
    
47.             } catch (InterruptedException ie) {  
    
48.                 // On interruption, re-check runState  
    
49.             }  
    
50.         }  
    
51.     }  
    
52.     }  
    
53.     ..........  
    
54. }

復制代碼

Worker是一個內(nèi)部類。根據(jù)之前可以知道，傳入addThread的參數(shù)是null，也就是說Work中firstTask為null。
        在看看newThread是一個什么方法：

1. public class Executors {  
   
2.     ..........  
   
3.     static class DefaultThreadFactory implements ThreadFactory {  
   
4.         ..........  
   
5.         public Thread newThread(Runnable r) {  
   
6.             Thread t = new Thread(group, r,  
   
7.                                   namePrefix + threadNumber.getAndIncrement(),  
   
8.                                   0);  
   
9.             if (t.isDaemon())  
   
10.                 t.setDaemon(false);  
    
11.             if (t.getPriority() != Thread.NORM_PRIORITY)  
    
12.                 t.setPriority(Thread.NORM_PRIORITY);  
    
13.             return t;  
    
14.         }  
    
15.         ..........  
    
16.     }  
    
17.     ..........  
    
18. }  

復制代碼

通過源碼可以得知threadFactory的實際類型是DefaultThreadFactory，而DefaultThreadFactory是Executors的一個嵌套內(nèi)部類。

        之前我們提到了t.start()這個方法執(zhí)行了線程。那么現(xiàn)在從頭順一下，看看到底是執(zhí)行了誰的run方法。首先知道，t=addThread(null)，而addThread內(nèi)部執(zhí)行了下面三步，Worker w = new Worker(null);Thread t = threadFactory.newThread(w);return t;這里兩個t是一致的。
        從這里可以看出，t.start()實際上執(zhí)行的是Worker內(nèi)部的run方法。run()內(nèi)部會在if條件里面使用“短路”：判斷firstTask是否為null，若不是null則直接執(zhí)行firstTask的run方法；如果是null，則調(diào)用getTask()方法來獲取Runnable類型實例。從哪里獲取呢？workQueue！在execute方法中，執(zhí)行ensureQueuedTaskHandled(command)之前就已經(jīng)把Runnable類型實例放入到workQueue中了，所以這里可以從workQueue中獲取到。