資訊專欄INFORMATION COLUMN
摘要：能夠異步的執(zhí)行任務(wù)，并且通常管理一個線程池。這樣我們就不用手動的去創(chuàng)建線程了，線程池中的所有線程都將被重用。在之后不能再提交任務(wù)到線程池。它不使用固定大小的線程池，默認(rèn)情況下是主機(jī)的可用內(nèi)核數(shù)。
原文地址： Java 8 Concurrency Tutorial: Threads and Executors
Java 5 初次引入了Concurrency API，并在隨后的發(fā)布版本中不斷優(yōu)化和改進(jìn)。這篇文章的大部分概念也適用于老的版本。我的代碼示例主要聚焦在Java 8上，并大量適用 lambda 表達(dá)式和一些新特性。如果你還不熟悉 lambda 表達(dá)式，建議先閱讀 Java 8 Tutorial。
Threads 和 Runnables所有現(xiàn)代操作系統(tǒng)都是通過進(jìn)程和線程來支持并發(fā)的。進(jìn)程通常是相互獨(dú)立運(yùn)行的程序?qū)嵗?。例如，你啟動一個 Java 程序，操作系統(tǒng)會產(chǎn)生一個新的進(jìn)程和其他程序并行運(yùn)行。在這些進(jìn)程中可以利用線程同時執(zhí)行代碼。這樣我們就可以充分利用 CPU。
Java 從 JDK 1.0 開始就支持線程。在開始一個新線程之前，必須先指定運(yùn)行的代碼，通常稱為 Task。下面是通過實(shí)現(xiàn) Runnable 接口來啟動一個新線程的例子：
Runnable task = () -> {
String threadName = Thread.currentThread().getName();
System.out.println("Hello " + threadName);
};
task.run();
Thread thread = new Thread(task);
thread.start();
System.out.println("Done!");
由于 Runnable 是一個函數(shù)式接口，我們可以使用 lambda 表達(dá)式來打印線程的名字到控制臺。我們直接在主線程上執(zhí)行Runnable，然后開始一個新線程。在控制臺你將看到這樣的結(jié)果：
Hello main Hello Thread-0 Done!
或者：
Hello main Done! Hello Thread-0
由于是并發(fā)執(zhí)行，我們無法預(yù)測 Runnable 是在打印 Done 之前還是之后調(diào)用，順序不是不確定的，因此并發(fā)編程成為大型應(yīng)用程序開發(fā)中一項(xiàng)復(fù)雜的任務(wù)。
線程也可以休眠一段時間，例如下面的例子：
Runnable runnable = () -> {
try {
String name = Thread.currentThread().getName();
System.out.println("Foo " + name);
TimeUnit.SECONDS.sleep(1);
System.out.println("Bar " + name);
}
catch (InterruptedException e) {
e.printStackTrace();
}
};
Thread thread = new Thread(runnable);
thread.start();
執(zhí)行上面的代碼會在兩個打印語句之間停留1秒鐘。TimeUnit 是一個時間單位的枚舉，或者可以通過調(diào)用 Thread.sleep(1000) 實(shí)現(xiàn)。
使用 Thread 類可能非常繁瑣且容易出錯。由于這個原因，在2004年，Java 5版本引入了 Concurrency API。API 位于 java.util.concurrent 包下，包含了許多有用的有關(guān)并發(fā)編程的類。從那時起，每個新發(fā)布的 Java 版本都增加了并發(fā) API，Java 8 也提供了新的類和方法來處理并發(fā)。
現(xiàn)在我們來深入了解一下Concurrency API中最重要的部分 - executor services。
ExecutorsConcurrency API 引入了 ExecutorService 的概念，作為處理線程的高級別方式用來替代 Threads。 Executors 能夠異步的執(zhí)行任務(wù)，并且通常管理一個線程池。這樣我們就不用手動的去創(chuàng)建線程了，線程池中的所有線程都將被重用。從而可以在一個
executor service 的整個應(yīng)用程序生命周期中運(yùn)行盡可能多的并發(fā)任務(wù)。
下面是一個簡單的 executors 例子：
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(() -> {
String threadName = Thread.currentThread().getName();
System.out.println("Hello " + threadName);
});
// => Hello pool-1-thread-1
Executors 類提供了方便的工廠方法來創(chuàng)建不同類型的 executor services 。在這個例子中使用了只執(zhí)行一個線程的 executor。
執(zhí)行結(jié)果看起來和上面的示例類似，但是你會注意到一個重要區(qū)別：Java 進(jìn)程永遠(yuǎn)不會停止，執(zhí)行者必須明確的停止它，否則它會不斷的接受新的任務(wù)。
ExecutorService 為此提供了兩種方法：shutdown() 等待當(dāng)前任務(wù)執(zhí)行完畢，而 shutdownNow() 則中斷所有正在執(zhí)行的任務(wù)，并立即關(guān)閉執(zhí)行程序。在 shudown 之后不能再提交任務(wù)到線程池。
下面是我關(guān)閉程序的首選方式：
try {
System.out.println("attempt to shutdown executor");
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
System.err.println("tasks interrupted");
}
finally {
if (!executor.isTerminated()) {
System.err.println("cancel non-finished tasks");
}
executor.shutdownNow();
System.out.println("shutdown finished");
}
執(zhí)行者調(diào)用 shutdown 關(guān)閉 executor，在等待 5 秒鐘鐘后，不管任務(wù)有沒有執(zhí)行完畢都調(diào)用 shutdownNow 中斷正在執(zhí)行的任務(wù)而關(guān)閉。
Callables 和 Futures除了 Runnable 以外，executors 還支持 Callable 任務(wù)，和 Runnable 一樣是一個函數(shù)式接口，但它是有返回值的。
下面是一個使用 lambda 表達(dá)式定義的 Callable ，在睡眠 1 秒后返回一個整形值。
Callabletask = () -> { try { TimeUnit.SECONDS.sleep(1); return 123; } catch (InterruptedException e) { throw new IllegalStateException("task interrupted", e); } };
和 Runnable 一樣，Callable 也可以提交到 executor services，但是執(zhí)行的結(jié)果是什么？由于 submit() 不等待任務(wù)執(zhí)行完成，executor service 不能直接返回調(diào)用的結(jié)果。相對應(yīng)的，它返回一個 Future 類型的結(jié)果，使用 Future 可以檢索實(shí)際執(zhí)行結(jié)果。
ExecutorService executor = Executors.newFixedThreadPool(1); Futurefuture = executor.submit(task); System.out.println("future done? " + future.isDone()); Integer result = future.get(); System.out.println("future done? " + future.isDone()); System.out.print("result: " + result);
在將 Callable 提交給 executor 后，首先通過 isDone() 來檢查 future 是否執(zhí)行完畢。我敢肯定，情況并非如此，因?yàn)樯厦娴恼{(diào)用在返回整數(shù)之前睡眠了 1 秒鐘。
調(diào)用方法 get() 會阻塞當(dāng)前線程，直到 callable 執(zhí)行完成返回結(jié)果，現(xiàn)在 future 執(zhí)行完成，并在控制臺輸出下面的結(jié)果：
future done? false future done? true result: 123
Future 與 executor service 緊密結(jié)合，如果關(guān)閉 executor service， 每個 Future 都會拋出異常。
executor.shutdownNow(); future.get();
這里創(chuàng)建 executor 的方式與前面的示例不同，這里使用 newFixedThreadPool(1) 來創(chuàng)建一個線程數(shù)量為 1 的線程池來支持 executor， 這相當(dāng)于 newSingleThreadExecutor() ，稍后我們我們會通過傳遞一個大于 1 的值來增加線程池的大小。
Timeouts任何對 future.get()的調(diào)用都會阻塞并等待 Callable 被終止。 在最壞的情況下，一個可調(diào)用函數(shù)將永遠(yuǎn)運(yùn)行，從而使應(yīng)用程序無法響應(yīng)?？梢院唵蔚赝ㄟ^超時來抵消這些情況：
ExecutorService executor = Executors.newFixedThreadPool(1); Futurefuture = executor.submit(() -> { try { TimeUnit.SECONDS.sleep(2); return 123; } catch (InterruptedException e) { throw new IllegalStateException("task interrupted", e); } }); future.get(1, TimeUnit.SECONDS);
執(zhí)行上面的代碼會拋出 TimeoutException
Exception in thread "main" java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask.get(FutureTask.java:205)
指定了 1 秒鐘的最長等待時間，但是在返回結(jié)果之前，可調(diào)用事實(shí)上需要 2 秒鐘的時間。
InvokeAllExecutors 支持通過 invokeAll() 批量提交多個 Callable 。這個方法接受一個 Callable 類型集合的參數(shù)，并返回一個 Future 類型的 List。
