Java中使用ThreadPoolExecutor並行執行獨立的單執行緒任務的詳細介紹-java教程-PHP中文網

Java SE 5.0中引入了任務執行框架，這是簡化多執行緒程式設計開發的一大進步。使用這個框架可以方便地管理任務：管理任務的生命週期以及執行策略。

在這篇文章中，我們透過一個簡單的例子來展現這個框架所帶來的靈活與簡單。

基礎

執行框架引入了Executor介面來管理任務的執行。 Executor是一個用來提交Runnable任務的介面。這個介面將任務提交與任務執行隔離起來：擁有不同執行策略的executor都實作了同一個提交介面。改變執行策略不會影響任務的提交邏輯。

如果你要提交一個Runnable物件來執行，很簡單：

Executor exec = …;
exec.execute(runnable);

登入後複製

執行緒池

如前所述，executor如何去執行提交的runnable任務並沒有在Executor介面中規定，這取決於你所使用的executor的具體類型。這個框架提供了幾種不同的executor，執行策略針對不同的場景而不同。

你可能會用到的最常見的executor類型就是線程池executor，也就是ThreadPoolExecutor類別（及其子類別）的實例。 ThreadPoolExecutor管理一個執行緒池和一個工作佇列，執行緒池存放著用於執行任務的工作執行緒。

你肯定在其他技術中也了解「池」的概念。使用「池」的一個最大的好處是減少資源創建的開銷，用過並釋放後，還可以重複使用。另一個間接的好處是你可以控制使用資源的多寡。例如，你可以調整線程池的大小達到你想要的負載，而不損害系統的資源。

這個框架提供了一個工廠類，叫做Executors，來創建線程池。使用這個工程類別你可以建立不同特性的執行緒池。儘管底層的實作常常是一樣的（ThreadPoolExecutor），但工廠類別可以使你不必使用複雜的建構子就可以快速地設定一個執行緒池。工程類別的工廠方法有：

newFixedThreadPool：此方法傳回一個最大容量固定的執行緒池。它會按需建立新線程，線程數量不大於配置的數量大小。當執行緒數達到最大以後，執行緒池會一直維持這麼多不變。
newCachedThreadPool：此方法傳回一個無界的執行緒池，也就是沒有最大數量限制。但當工作量減小時，這類執行緒池會銷毀沒用的執行緒。
newSingleThreadedExecutor：該方法傳回一個executor，它可以保證所有的任務都在一個單執行緒中執行。
newScheduledThreadPool：此方法傳回一個固定大小的執行緒池，它支援延時和定時任務的執行。

這只是一個開端。 Executor還有一些其他用法已經超出了這篇文章的範圍，我強烈建議你研究以下內容：

生命週期管理的方法，這些方法由ExecutorService介面聲明（例如shutdown ()和awaitTermination()）。
使用CompletionService來查詢任務狀態、取得回傳值，如果有回傳值的話。

ExecutorService介面特別重要，因為它提供了關閉執行緒池的方法，並確保清理了不再使用的資源。令人欣慰的是，ExecutorService介面相當簡單、一目了然，我建議全面地學習下它的文件。

大致來說，當你向ExecutorService發送了一個shutdown()訊息後，它就不會接收新提交的任務，但是仍在佇列中的任務會被繼續處理完。你可以使用isTerminated()來查詢ExecutorService終止狀態，或使用awaitTermination(…)方法來等待ExecutorService終止。如果傳入一個最大超時時間作為參數，awaitTermination方法就不會永遠等待。

警告： 對JVM進程永遠不會退出的理解上，存在著一些錯誤和迷惑。如果你不關閉executorService，只是銷毀了底層的線程，JVM就不會退出。當最後一個普通線程（非守護線程）退出後，JVM也會退出。

配置ThreadPoolExecutor

如果你決定不使用Executor的工廠類，而是手動建立一個 ThreadPoolExecutor，你需要使用建構子來建立並配置。以下是這個類別使用最廣泛的一個建構函數：

public ThreadPoolExecutor(
    int corePoolSize,
    int maxPoolSize,
    long keepAlive,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    RejectedExecutionHandler handler);

登入後複製

如你所見，你可以配置以下內容：

核心池的大小（執行緒池將會使用的大小）
最大池大小
#存活時間，空閒執行緒在這個時間後被銷毀
存放任務的工作佇列
任務提交拒絕後要執行的策略

限制队列中任务数

限制执行任务的并发数、限制线程池大小对应用程序以及程序执行结果的可预期性与稳定性有很大的好处。无尽地创建线程，最终会耗尽运行时资源。你的应用程序因此会产生严重的性能问题，甚至导致程序不稳定。

这只解决了部分问题：限制了并发任务数，但并没有限制提交到等待队列的任务数。如果任务提交的速率一直高于任务执行的速率，那么应用程序最终会出现资源短缺的状况。

解决方法是：

为Executor提供一个存放待执行任务的阻塞队列。如果队列填满，以后提交的任务会被“拒绝”。
当任务提交被拒绝时会触发RejectedExecutionHandler，这也是为什么这个类名中引用动词“rejected”。你可以实现自己的拒绝策略，或者使用框架内置的策略。

默认的拒绝策略可以让executor抛出一个RejectedExecutionException异常。然而，还有其他的内建策略：

悄悄地丢弃一个任务
丢弃最旧的任务，重新提交最新的
在调用者的线程中执行被拒绝的任务

什么时候以及为什么我们才会这样配置线程池？让我们看一个例子。

示例：并行执行独立的单线程任务

最近，我被叫去解决一个很久以前的任务的问题，我的客户之前就运行过这个任务。大致来说，这个任务包含一个组件，这个组件监听目录树所产生的文件系统事件。每当一个事件被触发，必须处理一个文件。一个专门的单线程执行文件处理。说真的，根据任务的特点，即使我能把它并行化，我也不想那么做。一天的某些时候，事件到达率才很高，文件也没必要实时处理，在第二天之前处理完即可。

当前的实现采用了一些混合且匹配的技术，包括使用UNIX SHELL脚本扫描目录结构，并检测是否发生改变。实现完成后，我们采用了双核的执行环境。同样，事件的到达率相当低：目前为止，事件数以百万计，总共要处理1～2T字节的原始数据。

运行处理程序的主机是12核的机器：很好机会去并行化这些旧的单线程任务。基本上，我们有了食谱的所有原料，我们需要做的仅仅是把程序建立起来并调节。在写代码前，我们必须了解下程序的负载。我列一下我检测到的内容：

有非常多的文件需要被周期性地扫描：每个目录包含1～2百万个文件
扫描算法很快，可以并行化
处理一个文件至少需要1s，甚至上升到2s或3s
处理文件时，性能瓶颈主要是CPU
CPU利用率必须可调，根据一天时间的不同而使用不同的负载配置。

我需要这样一个线程池，它的大小在程序运行的时候通过负载配置来设置。我倾向于根据负载策略创建一个固定大小的线程池。由于线程的性能瓶颈在CPU，它的核心使用率是100%，不会等待其他资源，那么负载策略就很好计算了：用执行环境的CPU核心数乘以一个负载因子（保证计算的结果在峰值时至少有一个核心）：

int cpus = Runtime.getRuntime().availableProcessors();
int maxThreads = cpus * scaleFactor;
maxThreads = (maxThreads > 0 ? maxThreads : 1);

登入後複製

然后我需要使用阻塞队列创建一个ThreadPoolExecutor，可以限制提交的任务数。为什么？是这样，扫描算法执行很快，很快就产生庞大数量需要处理的文件。数量有多庞大呢？很难预测，因为变动太大了。我不想让executor内部的队列不加选择地填满了要执行的任务实例（这些实例包含了庞大的文件描述符）。我宁愿在队列填满时，拒绝这些文件。

而且，我将使用ThreadPoolExecutor.CallerRunsPolicy作为拒绝策略。为什么？因为当队列已满时，线程池的线程忙于处理文件，我让提交任务的线程去执行它（被拒绝的任务）。这样，扫面会停止，转而去处理一个文件，处理结束后马上又会扫描目录。

下面是创建executor的代码：

ExecutorService executorService =
    new ThreadPoolExecutor(
        maxThreads, // core thread pool size
        maxThreads, // maximum thread pool size
        1, // time to wait before resizing pool
        TimeUnit.MINUTES, 
        new ArrayBlockingQueue<Runnable>(maxThreads, true),
        new ThreadPoolExecutor.CallerRunsPolicy());

登入後複製

下面是程序的框架（极其简化版）：

// scanning loop: fake scanning
while (!dirsToProcess.isEmpty()) {
    File currentDir = dirsToProcess.pop();

    // listing children
    File[] children = currentDir.listFiles();

    // processing children
    for (final File currentFile : children) {
        // if it&#39;s a directory, defer processing
        if (currentFile.isDirectory()) {
            dirsToProcess.add(currentFile);
            continue;
        }

        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    // if it&#39;s a file, process it
                    new ConvertTask(currentFile).perform();
                } catch (Exception ex) {
                    // error management logic
                }
            }
        });
    }
}

// ...
// wait for all of the executor threads to finish
executorService.shutdown();
try {
    if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
        // pool didn&#39;t terminate after the first try
        executorService.shutdownNow();
    }

    if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
        // pool didn&#39;t terminate after the second try
    }
} catch (InterruptedException ex) {
    executorService.shutdownNow();
    Thread.currentThread().interrupt();
}

登入後複製