介紹

隨著當(dāng)今處理器中可用的核心數(shù)量的增加， 隨著對實現(xiàn)更高吞吐量的需求的不斷增長，多線程 API 變得非常流行。 Java 提供了自己的多線程框架，稱為 Executor 框架.

1. Executor 框架是什么？

Executor 框架包含一組用于有效管理工作線程的組件。Executor API 通過 Executors 將任務(wù)的執(zhí)行與要執(zhí)行的實際任務(wù)解耦。 這是 生產(chǎn)者-消費(fèi)者 模式的一種實現(xiàn)。

java.util.concurrent.Executors 提供了用于創(chuàng)建工作線程的線程池的工廠方法。

為了使用 Executor 框架，我們需要創(chuàng)建一個線程池并提交任務(wù)給它以供執(zhí)行。Executor 框架的工作是調(diào)度和執(zhí)行已提交的任務(wù)并從線程池中拿到返回的結(jié)果。

浮現(xiàn)于腦海中的一個基本的問題是，當(dāng)我們創(chuàng)建 java.lang.Thread 的對象或調(diào)用實現(xiàn)了 Runnable/Callable 接口來達(dá)到的程序的并行性時，為什么需要線程池？

答案來源于兩個基本面：

• 為新任務(wù)創(chuàng)建新的線程會存在額外的線程創(chuàng)建以及銷毀的開銷。管理這些線程的生命周期會明顯增加 CPU 的執(zhí)行時間。
  
• 不進(jìn)行任何限制地為每個進(jìn)程創(chuàng)建線程會導(dǎo)致創(chuàng)建大量線程。這些線程會占用大量內(nèi)存并引起資源的浪費(fèi)。當(dāng)一個線程利用完 CPU 的時間片后另一個線程即將利用CPU的時間片時，CPU 會花費(fèi)大量的時間來切換線程的上下文。

所有的這些因素都會導(dǎo)致系統(tǒng)的吞吐量下降。線程池通過保持線程一直存活并重用這些線程來克服這個問題。當(dāng)提交到線程池中的任務(wù)多于正在執(zhí)行的線程時，那些多余的任務(wù)將被放到隊列中。 一旦執(zhí)行任務(wù)的線程有空閑的了，它們會從隊列中取下一個任務(wù)來執(zhí)行。對于 JDK 提供的現(xiàn)成的 executors 此任務(wù)隊列基本是無界的。

2. Executors 的類型

現(xiàn)在我們已經(jīng)了解了 executors 是什么， 讓我們來看看不同類型的 executors。

2.1 SingleThreadExecutor

此線程池 executor 只有一個線程。它用于以順序方式的形式執(zhí)行任務(wù)。如果此線程在執(zhí)行任務(wù)時因異常而掛掉，則會創(chuàng)建一個新線程來替換此線程，后續(xù)任務(wù)將在新線程中執(zhí)行。

ExecutorService executorService = Executors.newSingleThreadExecutor()

2.2 FixedThreadPool(n)

顧名思義，它是一個擁有固定數(shù)量線程的線程池。提交給 executor 的任務(wù)由固定的 n 個線程執(zhí)行，如果有更多的任務(wù)，它們存儲在 LinkedBlockingQueue 里。這個數(shù)字 n 通常跟底層處理器支持的線程總數(shù)有關(guān)。

ExecutorService executorService = Executors.newFixedThreadPool(4);

2.3 CachedThreadPool

該線程池主要用于執(zhí)行大量短期并行任務(wù)的場景。與固定線程池不同，此線程池的線程數(shù)不受限制。如果所有的線程都在忙于執(zhí)行任務(wù)并且又有新的任務(wù)到來了，這個線程池將創(chuàng)建一個新的線程并將其提交到 executor。只要其中一個線程變?yōu)榭臻e，它就會執(zhí)行新的任務(wù)。 如果一個線程有 60 秒的時間都是空閑的，它們將被結(jié)束生命周期并從緩存中刪除。

但是，如果管理得不合理，或者任務(wù)不是很短的，則線程池將包含大量的活動線程。這可能導(dǎo)致資源紊亂并因此導(dǎo)致性能下降。

ExecutorService executorService = Executors.newCachedThreadPool();

2.4 ScheduledExecutor

當(dāng)我們有一個需要定期運(yùn)行的任務(wù)或者我們希望延遲某個任務(wù)時，就會使用此類型的 executor。

ScheduledExecutorService scheduledExecService = Executors.newScheduledThreadPool(1);

可以使用 scheduleAtFixedRate 或 scheduleWithFixedDelay 在 ScheduledExecutor 中定期的執(zhí)行任務(wù)。

scheduledExecService.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

scheduledExecService.scheduleWithFixedDelay(Runnable command, long initialDelay, long period, TimeUnit unit)

這兩種方法的主要區(qū)別在于它們對連續(xù)執(zhí)行定期任務(wù)之間的延遲的應(yīng)答。

• scheduleAtFixedRate：無論前一個任務(wù)何時結(jié)束，都以固定間隔執(zhí)行任務(wù)。
• scheduleWithFixedDelay：只有在當(dāng)前任務(wù)完成后才會啟動延遲倒計時。

3. 對于 Future 對象的理解

可以使用 executor 返回的 java.util.concurrent.Future 對象訪問提交給 executor 的任務(wù)的結(jié)果。 Future 可以被認(rèn)為是 executor 對調(diào)用者的響應(yīng)。

Future<String> result = executorService.submit(callableTask);

如上所述，提交給 executor 的任務(wù)是異步的，即程序不會等待當(dāng)前任務(wù)執(zhí)行完成，而是直接進(jìn)入下一步。相反，每當(dāng)任務(wù)執(zhí)行完成時，executor 在此 Future對象中設(shè)置它。

調(diào)用者可以繼續(xù)執(zhí)行主程序，當(dāng)需要提交任務(wù)的結(jié)果時，他可以在這個 Future對象上調(diào)用.get() 方法來獲取。如果任務(wù)完成，結(jié)果將立即返回給調(diào)用者，否則調(diào)用者將被阻塞，直到 executor 完成此操作的執(zhí)行并計算出結(jié)果。

如果調(diào)用者不能無限期地等待任務(wù)執(zhí)行的結(jié)果，那么這個等待時間也可以設(shè)置為定時地?？梢酝ㄟ^ Future.get(long timeout，TimeUnit unit) 方法實現(xiàn)，如果在規(guī)定的時間范圍內(nèi)沒有返回結(jié)果，則拋出 TimeoutException。調(diào)用者可以處理此異常并繼續(xù)執(zhí)行該程序。

如果在執(zhí)行任務(wù)時出現(xiàn)異常，則對 get 方法的調(diào)用將拋出一個ExecutionException。

對于 Future.get()方法返回的結(jié)果，一個重要的事情是，只有提交的任務(wù)實現(xiàn)了java.util.concurrent.Callable接口時才返回 Future。如果任務(wù)實現(xiàn)了Runnable接口，那么一旦任務(wù)完成，對 .get() 方法的調(diào)用將返回 null。

另一個關(guān)注點(diǎn)是 Future.cancel(boolean mayInterruptIfRunning) 方法。此方法用于取消已提交任務(wù)的執(zhí)行。如果任務(wù)已在執(zhí)行，則 executor 將嘗試在mayInterruptIfRunning 標(biāo)志為 true 時中斷任務(wù)執(zhí)行。

4. Example: 創(chuàng)建和執(zhí)行一個簡單的 Executor

我們現(xiàn)在將創(chuàng)建一個任務(wù)并嘗試在 fixed pool executor 中執(zhí)行它：

public class Task implements Callable<String> {

  private String message;

  public Task(String message) {
    this.message = message;
  }

  @Override
  public String call() throws Exception {
    return "Hello " + message + "!";
  }
}

Task 類實現(xiàn) Callable 接口并有一個 String 類型作為返回值的方法。 這個方法也可以拋出 Exception。這種向 executor 拋出異常的能力以及 executor 將此異常返回給調(diào)用者的能力非常重要，因為它有助于調(diào)用者知道任務(wù)執(zhí)行的狀態(tài)。

現(xiàn)在讓我們來執(zhí)行一下這個任務(wù)：

public class ExecutorExample { 
  public static void main(String[] args) {

    Task task = new Task("World");

    ExecutorService executorService = Executors.newFixedThreadPool(4);
    Future<String> result = executorService.submit(task);

    try {
      System.out.println(result.get());
    } catch (InterruptedException | ExecutionException e) {
      System.out.println("Error occured while executing the submitted task");
      e.printStackTrace();
    }

    executorService.shutdown();
  }
}

我們創(chuàng)建了一個具有4個線程數(shù)的 FixedThreadPool executors，因為這個 demo是在四核處理器上開發(fā)的。如果正在執(zhí)行的任務(wù)執(zhí)行大量 I/O 操作或花費(fèi)較長時間等待外部資源，則線程數(shù)可能超過處理器的核心數(shù)。

我們實例化了 Task 類，并將它提交給 executors 執(zhí)行。 結(jié)果由 Future 對象返回，然后我們在屏幕上打印。

讓我們運(yùn)行 ExecutorExample 并查看其輸出：

Hello World!

正如所料，任務(wù)追加了問候語 Hello 并通過 Future object 返回結(jié)果。

最后，我們調(diào)用 executorService 對象上的 shutdown 來終止所有線程并將資源返回給 OS。

.shutdown() 方法等待 executor 完成當(dāng)前提交的任務(wù)。 但是，如果要求是立即關(guān)閉 executor 而不等待，那么我們可以使用 .shutdownNow() 方法。

任何待執(zhí)行的任務(wù)都將結(jié)果返回到 java.util.List 對象中。

我們也可以通過實現(xiàn) Runnable 接口來創(chuàng)建同樣的任務(wù)：

public class Task implements Runnable{

  private String message;

  public Task(String message) {
    this.message = message;
  }

  public void run() {
    System.out.println("Hello " + message + "!");
  }
}

當(dāng)我們實現(xiàn) Runnable 時，這里有一些重要的變化。

無法從 run() 方法得到任務(wù)執(zhí)行的結(jié)果。 因此，我們直接在這里打印。

run() 方法不可拋出任何已受檢的異常。

5. 總結(jié)

隨著處理器時鐘速度難以提高，多線程正變得越來越主流。 但是，由于涉及復(fù)雜性，處理每個線程的生命周期非常困難。

在本文中，我們展示了一個高效而簡單的多線程框架，即 Executor Framework，并解釋了它的不同組件。 我們還看了一下在 executor 中創(chuàng)建提交和執(zhí)行任務(wù)的不同示例。

與往常一樣，此示例的代碼可以在GitHub上找到。

英文原文：https://stackabuse.com/concurrency-in-java-the-executor-framework/

最新評論