## Java專題十三（1）：線程 [TOC] ### 13.1.1.線程的定義 **進程是資源分配的最小單位，線程是CPU調度的最小單位** 電腦上開一個QQ應用程序就是一個進程，你用QQ與**甲**打電話的同時跟與**乙**文字聊天，打電話和文字聊天都是一個線程，這里使用2個線程來完成用戶不同的操作，一個QQ進程包管理著多個線程 **線程生命周期**： 1. 新建（New）狀態，當程序使用new關鍵字創建了一個線程之后，該線程就處于新建狀態，此時僅由JVM為其分配內存，并初始化其成員變量的值 2. 就緒（Runnable）狀態，當線程對象調用了start()方法之后，該線程處于就緒狀態。Java虛擬機會為其創建方法調用棧和程序計數器，等待調度運行 3. 運行（Running）狀態，如果處于就緒狀態的線程獲得了CPU，開始執行run()方法的線程執行體，則該線程處于運行狀態 4. 阻塞（Blocked）狀態，當處于運行狀態的線程失去所占用資源之后，便進入阻塞狀態 5. 死亡（Dead）狀態，線程出錯或者線程正常結束  ### 13.1.2.線程的創建 #### 13.1.2.1,實現Runnable接口 ``` class PrimeRun implements Runnable { long minPrime; PrimeRun(long minPrime) { this.minPrime = minPrime; } public void run() { // compute primes larger than minPrime } } PrimeRun p = new PrimeRun(143); new Thread(p).start(); ``` #### 13.1.2.2.繼承Thread類 ``` class PrimeThread extends Thread { long minPrime; PrimeThread(long minPrime) { this.minPrime = minPrime; } public void run() { // compute primes larger than minPrime } } PrimeThread p = new PrimeThread(143); p.start(); ``` #### 13.1.2.3.使用Callable和Future ``` class PrimeThread implements Callable<Long> { long minPrime; PrimeThread(long minPrime) { this.minPrime = minPrime; } public Long call() throws Exception { return this.minPrime; } } PrimeThread p = new PrimeThread(143); FutureTask task = new FutureTask(p); new Thread(task).start(); System.out.println(task.get()); ``` ### 13.1.3.線程池（ThreadPoolExecutor類） - **線程池的參數：** ~~~ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) ~~~ **corePoolSize**：保存在線程池的線程數量，即使它們是空閑的 **maximumPoolSize**：允許加入到池中的最大線程數量 **keepAliveTime**：當線程數量比corePoolSize多時，為多余空閑線程終止之前等待新任務的最長時間 **workQueue**：線程池工作隊列 **threadFactory**：創建線程的工廠類 - **ThreadPoolExecutor其它參數** **workers**：工人集合 **workerCount**：有效線程的數量 **runState**：線程池的狀態，running, shutting down等 1. RUNNING ：接收新任務，處理已經入隊的任務 2. SHUTDOWN : 不接收新任務，但處理已經入隊的任務 3. STOP : 不接收新任務，不處理已經入隊的任務，并且中斷正在進行中的任務 4. TIDYING : 所有任務都完成 ，workerCount為零，該狀態下的線程會調用terminated()方法 5. TERMINATED : 完成方法terminated()的執行  為了更好理解線程池，先定義以下幾個概念： > 產品：實現了Runnable接口的線程類，用戶提交的Task任務 > 工人：ThreadPoolExecutor#Worker類，負責新產品的包裝工作和待包裝產品線的產品包裝工作 > 包裝：線程Runnable接口的實現類中run()方法實際執行的代碼 > 待包裝產品線：workQueue隊列，Task任務等待被執行 > 進口包裝機：核心線程池，數量為corePoolSize > 所有包裝機：最大線程池，數量為maximumPoolSize **=>線程池是怎么工作的呢？** 當有新的產品來時，首先查看進口包裝機是否都被占著(`workerCountOf(c) < corePoolSize`)，未被占著，就安排工人包裝新產品，都被占著的話，接著看待包裝產品線是否已滿(`workQueue.offer(command)`)，未滿的話加入新產品，等待包裝，否則，再看是否所有包裝機都被用完了(`addWorker(command, false)`)，沒有，則安排工人包裝新產品(`addWorker()`)，否則不包裝該產品(`reject(command)`)。  ```java // ThreadPoolExecutor#execute 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); } ``` **=>工人是怎么工作的?** 首先在`ThreadPoolExecutor#addWorker`方法添加一個`Worker`對象(實現了`Runnable`接口)，添加成功的話，啟動`w.thread`線程，`Worker`內部調用`runWorker`方法，在`runWorker`方法，一方面做新產品包裝工作(`task != null`->`task.run()`)，另一方面做待包裝產品線的產品包裝工作(`(task = getTask()) != null`->`task.run()`) ```java // ThreadPoolExecutor#addWorker private boolean addWorker(Runnable firstTask, boolean core) { boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // ... if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } // ThreadPoolExecutor#Worker private final class Worker extends AbstractQueuedSynchronizer implements Runnable { public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; // ... while (task != null || (task = getTask()) != null) { // ... task.run(); // ... } } } // ThreadPoolExecutor#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; } } } ```