Detaillierte Einführung in die Verwendung von ThreadPoolExecutor zum parallelen Ausführen unabhängiger Single-Threaded-Aufgaben in Java

Das Task Execution Framework wurde in Java SE 5.0 eingeführt, was einen großen Fortschritt bei der Vereinfachung der Multithread-Programmierentwicklung darstellt. Verwenden Sie dieses Framework, um Aufgaben einfach zu verwalten: Verwalten Sie den Lebenszyklus und die Ausführungsstrategie der Aufgabe.

In diesem Artikel verwenden wir ein einfaches Beispiel, um die Flexibilität und Einfachheit zu zeigen, die dieses Framework bietet.

Basic

Das Ausführungsframework führt die Executor-Schnittstelle ein, um die Ausführung von Aufgaben zu verwalten. Executor ist eine Schnittstelle zum Senden ausführbarer Aufgaben. Diese Schnittstelle isoliert die Aufgabenübermittlung von der Aufgabenausführung: Ausführende mit unterschiedlichen Ausführungsstrategien implementieren alle dieselbe Übermittlungsschnittstelle. Das Ändern der Ausführungsstrategie hat keine Auswirkungen auf die Aufgabenübermittlungslogik.

Wenn Sie ein ausführbares Objekt zur Ausführung übermitteln möchten, ist das ganz einfach:

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

Thread-Pool

Wie führt der Executor, wie oben erwähnt, das übermittelte Objekt aus? Ausführbare Aufgabe und nicht in der Executor-Schnittstelle angegeben. Dies hängt von der spezifischen Art des Executors ab, den Sie verwenden. Dieses Framework stellt mehrere unterschiedliche Executoren bereit und die Ausführungsstrategien variieren je nach Szenario.

Der am häufigsten verwendete Executortyp ist der Thread-Pool-Executor, der eine Instanz der ThreadPoolExecutor-Klasse (und ihrer Unterklassen) ist. ThreadPoolExecutor verwaltet einen Thread-Pool und eine Arbeitswarteschlange . Der Thread-Pool speichert die Arbeitsthreads, die zum Ausführen von Aufgaben verwendet werden.

Sie müssen das Konzept des „Pools“ in anderen Technologien verstanden haben. Einer der größten Vorteile der Verwendung eines „Pools“ besteht darin, dass die Kosten für die Ressourcenerstellung gesenkt werden. Nachdem er verwendet und freigegeben wurde, kann er wiederverwendet werden. Ein weiterer indirekter Vorteil besteht darin, dass Sie steuern können, wie viele Ressourcen verwendet werden. Sie können beispielsweise die Größe des Thread-Pools anpassen, um die gewünschte Auslastung zu erreichen, ohne die Systemressourcen zu beschädigen.

Dieses Framework stellt eine Factory-Klasse namens Executors zum Erstellen eines Thread-Pools bereit. Mit dieser Engineering-Klasse können Sie Thread-Pools mit unterschiedlichen Eigenschaften erstellen. Obwohl die zugrunde liegende Implementierung oft dieselbe ist (ThreadPoolExecutor), ermöglichen Ihnen Factory-Klassen die schnelle Einrichtung eines Thread-Pools, ohne komplexe

Konstruktoren zu verwenden. Die Factory-Methoden der Engineering-Klasse sind:

• newFixedThreadPool: Diese Methode gibt einen Thread-Pool mit einer festen maximalen Kapazität zurück. Es erstellt bei Bedarf neue Threads und die Anzahl der Threads ist nicht größer als die konfigurierte Anzahl. Wenn die Anzahl der Threads das Maximum erreicht, bleibt der Thread-Pool unverändert.

• newCachedThreadPool: Diese Methode gibt einen unbegrenzten Thread-Pool zurück, d. h. es gibt keine maximale Anzahlbegrenzung. Wenn jedoch die Arbeitslast abnimmt, zerstört diese Art von Thread-Pool ungenutzte Threads.

• newSingleThreadedExecutor: Diese Methode gibt einen Executor zurück, der sicherstellen kann, dass alle Aufgaben in einem einzigen Thread ausgeführt werden.

• newScheduledThreadPool: Diese Methode gibt einen Thread-Pool fester Größe zurück, der die Ausführung verzögerter und geplanter Aufgaben unterstützt.

Das ist erst der Anfang. Es gibt einige andere Verwendungsmöglichkeiten von Executor, die über den Rahmen dieses Artikels hinausgehen. Ich empfehle Ihnen dringend, Folgendes zu studieren:

• Lebenszyklusverwaltungsmethoden, die von der ExecutorService-Schnittstelle deklariert werden (z als „shutdown()“ und „awaitTermination()“.

• Verwenden Sie CompletionService, um den Aufgabenstatus abzufragen und den Rückgabewert abzurufen, sofern ein Rückgabewert vorhanden ist.

Die ExecutorService-Schnittstelle ist besonders wichtig, da sie eine Möglichkeit bietet, den Thread-Pool herunterzufahren und sicherzustellen, dass nicht mehr verwendete Ressourcen bereinigt werden. Glücklicherweise ist die ExecutorService-Schnittstelle recht einfach und selbsterklärend. Ich empfehle, die Dokumentation umfassend zu studieren.

Grob gesagt: Wenn Sie eine Shutdown()-Nachricht an ExecutorService senden, empfängt dieser keine neu übermittelten Aufgaben, aber Aufgaben, die sich noch in der Warteschlange befinden, werden weiterhin verarbeitet. Sie können isTerminated() verwenden, um den ExecutorService-Beendigungsstatus abzufragen, oder die Methode „awaitTermination(...)“ verwenden, um auf die Beendigung von ExecutorService zu warten. Wenn Sie als Parameter ein maximales Timeout übergeben, wartet die Methode „awaitTermination“ nicht ewig.

Warnung: Es gibt einige Fehler und Verwirrung im Verständnis, dass JVM-Prozesse niemals beendet werden. Wenn Sie den ExecutorService nicht schließen und einfach den zugrunde liegenden Thread zerstören, wird die JVM nicht beendet. Wenn der letzte normale Thread (Nicht-Daemon-Thread) beendet wird, wird auch die JVM beendet.

ThreadPoolExecutor konfigurieren

Wenn Sie sich entscheiden, nicht die Executor-Factory-Klasse zu verwenden, sondern manuell einen ThreadPoolExecutor zu erstellen, müssen Sie den Konstruktor verwenden, um ihn zu erstellen und zu konfigurieren. Hier ist einer der am häufigsten verwendeten Konstruktoren dieser Klasse:

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

Wie Sie sehen, können Sie Folgendes konfigurieren:

• Die Größe des Kerns Pool (Threads Die Größe, die der Pool verwenden wird)

• Maximale Poolgröße

• Die Überlebenszeit, inaktive Threads werden nach dieser Zeit zerstört

• Arbeitswarteschlange zum Speichern von Aufgaben

• Strategie, die ausgeführt werden soll, nachdem die Aufgabenübermittlung abgelehnt wurde

限制队列中任务数

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

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

解决方法是：

• 为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's a directory, defer processing
        if (currentFile.isDirectory()) {
            dirsToProcess.add(currentFile);
            continue;
        }

        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    // if it'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't terminate after the first try
        executorService.shutdownNow();
    }

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

总结

看到了吧，Java并发API非常简单易用，十分灵活，也很强大。真希望我多年前可以多花点功夫写一个这样简单的程序。这样我就可以在几小时内解决由传统单线程组件所引发的扩展性问题。

Das obige ist der detaillierte Inhalt vonDetaillierte Einführung in die Verwendung von ThreadPoolExecutor zum parallelen Ausführen unabhängiger Single-Threaded-Aufgaben in Java. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!