摘要：而且，線程池中的線程并沒(méi)有睡眠，而是進(jìn)入了自旋狀態(tài)。普通的線程被中斷會(huì)導(dǎo)致線程繼續(xù)執(zhí)行，從而方法運(yùn)行完畢，線程退出。線程死亡超過(guò)時(shí)間，任務(wù)對(duì)列沒(méi)有數(shù)據(jù)而返回。線程死亡保證了線程池至少留下個(gè)線程。

線程在執(zhí)行任務(wù)時(shí)，正常的情況是這樣的：

Thread  t=new Thread(new  Runnable() {            
            @Override
            public void run() {
                // TODO Auto-generated method stub    
            }
        });
        
        t.start();
        

??Thread 在初始化的時(shí)候傳入一個(gè)Runnable,以后就沒(méi)有機(jī)會(huì)再傳入一個(gè)Runable了。那么，woker作為一個(gè)已經(jīng)啟動(dòng)的線程。是如何不斷獲取Runnable的呢？
這個(gè)時(shí)候可以使用一個(gè)包裝器，將線程包裝起來(lái)，在Run方法內(nèi)部獲取任務(wù)。

public final class Worker implements Runnable {
    Thread thread = null;
    Runnable task;
    private BlockingQueue queues;
    public Worker(Runnable task, BlockingQueue queues) {
        this.thread = new Thread(this);
        this.task = task;
        this.queues = queues;
    }
    public void run() {
        if (task != null) {
            task.run();
        } 
            try {
                while (true) {
                    task = queues.take();
                    if (task != null) {
                        task.run();
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    public void start() {
        this.thread.start();
    }
}

public class Main {
    public static void main(String[] args) {
        BlockingQueue queues=new ArrayBlockingQueue(100);
        Worker  worker=new Worker(new Runnable() {
            public void run() {
                System.out.println("hello!!! ");
                try {
                    Thread.currentThread().sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }            
            }
        }, queues);
        worker.start();
        for(int i=0;i<100;i++){
            queues.offer(new Runnable() {
                public void run() {
                    System.out.println("hello!!! ");
                    try {
                        Thread.currentThread().sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

}

??這樣我們就簡(jiǎn)單地實(shí)現(xiàn)了一個(gè)“線程池”（可以將這個(gè)“線程池”改造成官方的模式，不過(guò)可以自己嘗試一下）。ThreadPool的這種實(shí)現(xiàn)模式是并發(fā)編程中經(jīng)典的Cyclic Work Distribution模式。
??那么，這種實(shí)現(xiàn)的線程池性能如何呢？
??由于其任務(wù)隊(duì)列使用的是阻塞隊(duì)列，在隊(duì)列內(nèi)部是自旋的。Reeteenlok是改進(jìn)的CLH隊(duì)列。自旋鎖會(huì)耗費(fèi)一定CPU的資源，在擁有大量任務(wù)執(zhí)行下的情況下比較有效。而且，線程池中的線程并沒(méi)有睡眠，而是進(jìn)入了自旋狀態(tài)。

??如果是不支持超線程的CPU，在同一時(shí)刻的確只能處理2個(gè)線程，但是并不意味著雙核的CPU只能處理兩個(gè)線程，它可以通過(guò)切換上下文來(lái)執(zhí)行多個(gè)線程。比如我只有一個(gè)大腦，但是我要處理5個(gè)人提交的任務(wù)，我可以處理完A的事情后，把事情的中間結(jié)果保存下，然后再處理B的，然后再讀取A的中間結(jié)果，處理A的事情。

??Woker自身繼承了Runnable,并對(duì)Thread做了一個(gè)包裝。Woker代碼如下所示：

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {

        private static final long serialVersionUID = 6138294804551838833L;

    
        Runnable firstTask;
   
        volatile long completedTasks;

 
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        public void run() {
            runWorker(this);
        }
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

??execute(Runnable command)方法內(nèi)部是這樣的：

public void execute(Runnable command) {
       if (command == null)
           throw new NullPointerException();
     
       int c = ctl.get();
       if (workerCountOf(c) < corePoolSize) {
           if (addWorker(command, true))
               return;
           c = ctl.get();
       }
       if (isRunning(c) && workQueue.offer(command)) {
           int recheck = ctl.get();
           if (! isRunning(recheck) && remove(command))
               reject(command);
           else if (workerCountOf(recheck) == 0)
               addWorker(null, false);
       }
       else if (!addWorker(command, false))
           reject(command);
   }

??ctl一個(gè)合并類(lèi)型的值。將當(dāng)前線程數(shù)和線程池狀態(tài)通過(guò)數(shù)學(xué)運(yùn)算合并到了一個(gè)值。具體是如何合并的可以參看一下源碼，這里就不敘述了。繼續(xù)向下走：

if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }

??可以看到，如果當(dāng)前線程數(shù)量小于了核心線程數(shù)量corePoolSize，就直接增加線程處理任務(wù)。與隊(duì)列沒(méi)有關(guān)系。但是緊接著又檢查了一遍狀態(tài)，因?yàn)樵谶@個(gè)過(guò)程中，別的線程也可能在添加任務(wù)。繼續(xù)向下走：

  if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        
        

??可以看到如果線程池是運(yùn)行態(tài)的，就把線程添加到任務(wù)隊(duì)列。workQueue是構(gòu)造函數(shù)傳遞過(guò)來(lái)的，可以是有界隊(duì)列，也可以是無(wú)界隊(duì)列。可以看出來(lái)，隊(duì)列如果是無(wú)界的，直接往隊(duì)列里面添加任務(wù)，這個(gè)時(shí)候，線程池中的線程也不會(huì)增加，一直會(huì)等于核心線程數(shù)。
??如果隊(duì)列是有界的，就嘗試直接新增線程處理任務(wù)，如果添加任務(wù)失敗，就調(diào)用reject方法來(lái)處理添加失敗的任務(wù)：

 else if (!addWorker(command, false))
            reject(command);

??來(lái)看看addWorker是如何實(shí)現(xiàn)的，邏輯流程已經(jīng)直接在注釋中說(shuō)明了。

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            //如果狀態(tài)大于SHUTDOWN，不再接受新的任務(wù)，直接返回
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

/**根據(jù)core來(lái)判斷，如果當(dāng)前線程數(shù)量大于corePoolSize或者最大線程數(shù)，直接返回。添加任務(wù)失敗。
**如果隊(duì)列是有界的或者任務(wù)添加到隊(duì)列失敗(參數(shù)core是false)，那么就會(huì)新開(kāi)一個(gè)線程處理業(yè)務(wù)，但如果線程已經(jīng)大于了maximumPoolSize，就會(huì)出現(xiàn)添加失敗，返回false。
*/

            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;

                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

??如果創(chuàng)建失敗的情況下會(huì)調(diào)用addWorkerFailed方法，從而將減少實(shí)際線程數(shù)。

??在addWorker中有這么一段代碼，表示為當(dāng)前線程數(shù)加1：

  private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }

??由于多線程可能同時(shí)操作。expect值可能會(huì)變化。僅僅一次的操作compareAndIncrementWorkerCount可能一次并不會(huì)成功，而且，一個(gè)線程在執(zhí)行addWork的過(guò)程中間，另外一個(gè)線程假設(shè)直接shotdown這個(gè)線程池。for循環(huán)的存在可以保證狀態(tài)一定是一致的。

在Worker中間實(shí)際上是調(diào)用的runWorker方法來(lái)執(zhí)行的具體業(yè)務(wù)：

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

??可以看到while循環(huán)不斷的從隊(duì)列中取出任務(wù)執(zhí)行。如果task==null 并且getTask()等于null的話，那么就會(huì)跳出循環(huán)，進(jìn)入到processWorkerExit，run方法執(zhí)行完畢以后，這個(gè)線程也被銷(xiāo)毀了。但是為什么在各自的線程執(zhí)行，為什么還需要加鎖呢？答案是因?yàn)橐€程池需要判斷這個(gè)線程是否在執(zhí)行任務(wù)。在interruptIdleWorkers方法中，要中斷那寫(xiě)目前空閑的線程，通過(guò)當(dāng)前Worker是否獲得了鎖就能判斷這個(gè)worker是否是空閑的:

 private void interruptIdleWorkers(boolean onlyOne) {
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try {
          for (Worker w : workers) {
              Thread t = w.thread;
              if (!t.isInterrupted() && w.tryLock()) {
                  try {
                      t.interrupt();
                  } catch (SecurityException ignore) {
                  } finally {
                      w.unlock(); //中斷不起作用。interrupt()對(duì)于自旋鎖是不起作用的。只是邏輯上被阻塞，
                  }
              }
              if (onlyOne)
                  break;
          }
      } finally {
          mainLock.unlock();
      }
  }

??可以看到，如果w.tryLock()可以獲取到鎖，那么就意味著當(dāng)前的 Woker并沒(méi)有處理任務(wù)（沒(méi)有進(jìn)入到循環(huán)里面或者被getTask方法所阻塞，無(wú)法獲取鎖）。
Work之所以繼承AbstractQueuedSynchronizer，而不去使用ReentrantLock。是因?yàn)镽eentrantLock是可重入鎖，在調(diào)用lock方法獲取鎖之后，再調(diào)用tryLock()還是會(huì)返回true。

public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        System.out.println(lock.tryLock());        
    }

輸出結(jié)果是true，所以使用ReentrantLock則難以判斷當(dāng)前Worker是否在執(zhí)行任務(wù)。

??在上面的interruptIdleWorkers方法中，線程被中斷。普通的線程被中斷會(huì)導(dǎo)致線程繼續(xù)執(zhí)行，從而run方法運(yùn)行完畢，線程退出。

對(duì)于一個(gè)沒(méi)有被阻塞的線程，中斷是不起作用的。中斷在如下線程被阻塞的方法中起作用：
the wait(),
wait(long),
wait(long, int)
join(),
join(long),
join(long, int),
sleep(long),
or sleep(long, int)
LockSupport.park(Object object);
LockSupport.park();

??，如果喚醒這些被阻塞的線程，從而能使得run方法繼續(xù)執(zhí)行，當(dāng)run方法執(zhí)行完畢，那么線程也就終結(jié)死亡。但是對(duì)于ReentrantLock和AbstractQueuedSynchronizer這種自旋+CAS實(shí)現(xiàn)的“邏輯鎖”，是不起作用的。
而且runWork本身也是While循環(huán)，靠中斷是無(wú)法退出循環(huán)的。

??但是在ThreadPoolExecutor的構(gòu)造函數(shù)中，有一個(gè)允許設(shè)置線程超時(shí)allowCoreThreadTimeOut參數(shù)的方法。如果允許超時(shí)，多于corePoolSize的線程將會(huì)在處在空閑狀態(tài)之后存活keepAliveTime時(shí)長(zhǎng)后終止。因此有了一個(gè)allowCoreThreadTimeOut方法：

 public void allowCoreThreadTimeOut(boolean value) {
        if (value && keepAliveTime <= 0)
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        if (value != allowCoreThreadTimeOut) {
            allowCoreThreadTimeOut = value;
            if (value)
                interruptIdleWorkers();
        }
    }

?? 正如上面提到的一樣，允許allowCoreThreadTimeOut并且調(diào)用interruptIdleWorkers方法并不能使線程退出。那么線程池又如何殺掉這個(gè)線程呢？
??沒(méi)錯(cuò)，就是getTask方法。只有當(dāng)getTask返回null的時(shí)候才能跳出While循環(huán)，run方法運(yùn)行完畢，那么線程自然而然就死亡了。getTask方法如下所示：

private Runnable getTask() {
      boolean timedOut = false; // Did the last poll() time out?
      for (;;) {
          int c = ctl.get();
          int rs = runStateOf(c);

          // Check if queue empty only if necessary.
          if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
              decrementWorkerCount();
              return null;
          }

          int wc = workerCountOf(c);
          // Are workers subject to culling?
          boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
          if ((wc > maximumPoolSize || (timed && timedOut))
              && (wc > 1 || workQueue.isEmpty())) {
              if (compareAndDecrementWorkerCount(c))
                  return null;
              continue;
          }
          try {
              Runnable r = timed ?
                  workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                  workQueue.take();
              if (r != null)
                  return r;
              timedOut = true;
          } catch (InterruptedException retry) {
              timedOut = false;
          }
      }
  }

??可以看到，如果線程池狀態(tài)大于SHUTDOWN并且隊(duì)列空，返回null，從而結(jié)束循環(huán)。（線程死亡）

??或者狀態(tài)大于SHUTDOWN并且線程大于STOP（STOP一定大于SHUTDOWN，所以可以直接說(shuō)線程大于STOP）返回null，從而結(jié)束循環(huán)。（線程死亡）
再往下可以看到如果超過(guò)了maximumPoolSize，返回null，從而結(jié)束循環(huán)。（線程死亡）
超過(guò)keepAliveTime時(shí)間，任務(wù)對(duì)列沒(méi)有數(shù)據(jù)而返回null。從而結(jié)束循環(huán)。（線程死亡）
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;保證了線程池至少留下corePoolSize個(gè)線程。

??在execute方法中，如果線程池拒絕添加任務(wù)，就會(huì)有一個(gè)鉤子方法來(lái)處理被拒絕的任務(wù)。
可以自己定義，也可以使用線城池中默認(rèn)的拒接處理協(xié)議。

AbortPolicy :直接拋出RejectedExecutionException異常；

CallerRunsPolicy：誰(shuí)調(diào)用的execute方法，誰(shuí)就執(zhí)行這個(gè)任務(wù)；

DiscardPolicy:直接丟棄，什么也不做；

DiscardOldestPolicy：丟棄對(duì)列中間最老的任務(wù)，執(zhí)行新任務(wù)。

有什么問(wèn)題或者建議，可以加入小密圈和我一起討論，或者在簡(jiǎn)書(shū)留言，歡迎喜歡和打賞。

最后向大家安利一本我寫(xiě)的關(guān)于Java并發(fā)的書(shū)籍：Java并發(fā)編程系統(tǒng)與模型，個(gè)人覺(jué)得寫(xiě)得不錯(cuò)，比較通俗易懂，非常適合初學(xué)者，百度閱讀可以下載電子書(shū)。