切入正題---------
可以在此處找到演示已接受答案的代碼：
完整示例：
https://github.com/NACHC-CAD/thread-example/tree/shutdown-first
執行：
https://github.com/NACHC-CAD/thread-example/blob/shutdown-first/src/main/java/com/nachc/examples/threadexample/WidgetFactory.java
原帖-------------------------
有很多使用 Java 執行緒和 Executors 的例子： https ://www.baeldung.com/thread-pool-java-and-guava
https://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
https://howtodoinjava.com/java/multi-threading/java-thread-pool-executor-example/
https://jenkov.com/tutorials/java-concurrency/thread-pools.html
https://xperti.io/blogs/thread-pools-java-introduction/
https://www.journaldev.com/1069/threadpoolexecutor-java-thread-pool-example-executorservice
https://stackify.com/java-thread-pools/
但是，我無法成功撰寫一個執行所有任務、等待任務完成然后正確終止的示例。
從這個例子開始作業：https ://howtodoinjava.com/java/multi-threading/java-thread-pool-executor-example/
該代碼僅呼叫 executor.shutdown()。如果執行緒消耗任何時間，這將不允許執行緒有時間完成。
我在這里創建了一個完整的最簡單的示例：https ://github.com/NACHC-CAD/thread-example/tree/await-termination
僅關閉分支涵蓋此用例（https://github.com/NACHC-CAD/thread-example/tree/shutdown-only）：
public void makeWidgets() {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(batchSize);
log.info("Building " howMany " widgets...");
for (int i = 0; i < howMany; i ) {
Widget widget = new Widget(lotNumber, i);
WidgetRunnable runnable = new WidgetRunnable(widget);
executor.execute(runnable);
}
log.info("SHUTTING DOWN----------------");
executor.shutdown();
}
此代碼給出以下輸出（應該創建 1000 個小部件，它們應該在等待 1 秒后報告它們已完成）。
2022-04-23 21:27:05,796 21:27:05.796 [main] INFO (WidgetFactoryIntegrationTest.java:12) - Starting test...
2022-04-23 21:27:05,799 21:27:05.799 [main] INFO (WidgetFactory.java:29) - Building 100 widgets...
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-2] INFO (Widget.java:24) - Starting build: 1/1
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-4] INFO (Widget.java:24) - Starting build: 1/3
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-1] INFO (Widget.java:24) - Starting build: 1/0
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-5] INFO (Widget.java:24) - Starting build: 1/4
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-6] INFO (Widget.java:24) - Starting build: 1/5
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-7] INFO (Widget.java:24) - Starting build: 1/6
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-8] INFO (Widget.java:24) - Starting build: 1/7
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-10] INFO (Widget.java:24) - Starting build: 1/9
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-9] INFO (Widget.java:24) - Starting build: 1/8
2022-04-23 21:27:05,801 21:27:05.801 [main] INFO (WidgetFactory.java:35) - SHUTTING DOWN----------------
2022-04-23 21:27:05,800 21:27:05.800 [pool-1-thread-3] INFO (Widget.java:24) - Starting build: 1/2
2022-04-23 21:27:05,801 21:27:05.801 [main] INFO (WidgetFactoryIntegrationTest.java:18) - Done.
如果我添加 executor.awaitTermination 代碼將運行所有執行緒但永遠不會終止。此示例在 await-termination 分支中：https ://github.com/NACHC-CAD/thread-example/tree/await-termination
public void makeWidgets() {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(batchSize);
log.info("Building " howMany " widgets...");
for (int i = 0; i < howMany; i ) {
Widget widget = new Widget(lotNumber, i);
WidgetRunnable runnable = new WidgetRunnable(widget);
executor.execute(runnable);
}
try {
executor.awaitTermination(1000, TimeUnit.HOURS);
} catch(Exception exp) {
throw(new RuntimeException(exp));
}
log.info("SHUTTING DOWN----------------");
executor.shutdown();
}
這段代碼讓所有的runnables完成但永遠不會退出。如何讓所有可運行物件完成并讓代碼運行完成（退出）？
uj5u.com熱心網友回復：
參考ThreadPoolExecutor檔案。awaitTermination() 方法描述如下：
在關閉請求后阻塞，直到所有任務都完成執行
而 shutdown() 方法描述讀取
啟動有序關閉，其中執行先前提交的任務，但不會接受新任務
這表明 awaitTermination() 呼叫在 shutdown() 呼叫后有效。
解決上述問題，需要先呼叫shutdown()，再呼叫awaitTermination()
注意：我沒有親自測驗過；但是，正如原始帖子的評論中所提到的那樣，約翰有，并且該機制有效
uj5u.com熱心網友回復：
Ironluca的答案是正確的。這是附加點和一些示例代碼。
一方面，不需要ThreadPoolExecutor直接宣告 & cast。只是使用比較通用的ExecutorService。
并且使用大小與您的批處理大小相匹配的執行緒池似乎是不明智的。在當前的 Java 中，您通常希望活動執行緒池少于 CPU 內核數。（如果Project Loom及其虛擬執行緒成功，這種計算方式將徹底改變，但這不是今天的現實，盡管您可以嘗試早期訪問構建。）
int threadPoolSize = 3 ; // Generally less than number of cores.
ExecutorService executorService = Executors.newFixedThreadPool( threadPoolSize );
讓我們簡化您的示例場景。我們定義Widget為一個簡單的記錄。
record Widget ( UUID id , Instant whenCreated ) {}
定義一個產生Widget. 我們想取回一個Widget物件，所以我們使用Callable而不是Runnable.
Callable < Widget > makeWidgetTask = ( ) -> {
Thread.sleep( Duration.ofMillis( 50 ).toMillis() ); // Pretend that we have a long-running task.
Widget widget = new Widget( UUID.randomUUID() , Instant.now() );
return widget;
};
做一個大集合，用于多次運行該任務。
List < Callable < Widget > > tasks = Collections.nCopies( 1_000 , makeWidgetTask );
實際上，我們需要封裝一個 try-catch。
List < Future < Widget > > futures = null;
try
{
futures = executorService.invokeAll( tasks );
}
catch ( InterruptedException e )
{
throw new RuntimeException( e );
}
將所有這些任務提交給執行器服務。注意我們如何取回Future物件串列。AFuture是我們對每個任務完成成功或失敗的句柄。
至于如何等待完成，如何使用ExecutorService#shutdown, shutdownNow, and awaitTermination，看Javadoc即可。?? 為您提供了樣板代碼的完整示例。
參考 Javadoc：
pool.shutdown(); // Disable new tasks from being submitted
try {
// Wait a while for existing tasks to terminate
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
pool.shutdownNow(); // Cancel currently executing tasks
// Wait a while for tasks to respond to being cancelled
if (!pool.awaitTermination(60, TimeUnit.SECONDS))
System.err.println("Pool did not terminate");
}
} catch (InterruptedException ex) {
// (Re-)Cancel if current thread also interrupted
pool.shutdownNow();
// Preserve interrupt status
Thread.currentThread().interrupt();
}
關鍵概念是shutdown不會停止任何正在進行的作業。當前正在執行的所有任務將繼續。當執行緒可用時，所有提交的任務最終將被安排在核心上執行。該shutdown方法只做一件事：停止任何進一步的任務被提交到這個執行器服務。參考 Javadoc：
shutdown()… 執??行之前提交的任務，但不會接受新任務。
進一步參考：
此方法不等待先前提交的任務完成執行。用來
awaitTermination做那個。
所以你需要在呼叫awaitTermination后呼叫shutdown。您在合理的時間內傳遞引數，您希望在該時間內完成、取消或中斷所有提交的任務。如果該時間限制已過，那么您可以假定出現問題。
請注意，呼叫shutdown不會阻塞，但呼叫會awaitTermination 阻塞。
讓我們將樣板代碼改編為我們自己的示例。
executorService.shutdown(); // Disable new tasks from being submitted.
try
{
if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
{
executorService.shutdownNow(); // Cancel currently executing tasks.
// Wait a while for tasks to respond to being cancelled.
if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
{ System.err.println( "Executor service did not terminate." ); }
}
}
catch ( InterruptedException ex )
{
executorService.shutdownNow(); // (Re-)Cancel if current thread also interrupted
Thread.currentThread().interrupt(); // Preserve interrupt status
}
最后，通過檢查Future物件的集合來回顧我們的結果。
System.out.println( "Count futures: " futures.size() );
for ( Future < Widget > future : futures )
{
if ( ! future.isDone() ) { System.out.println( "Oops! Task not done: " future.toString() ); }
else if ( future.isCancelled() ) { System.out.println( "Bummer. Task cancelled: " future.toString() ); }
else // Else task must have completed successfully.
{
try
{
Widget widget = future.get();
System.out.println( widget.toString() );
}
catch ( InterruptedException e )
{
throw new RuntimeException( e );
}
catch ( ExecutionException e )
{
throw new RuntimeException( e );
}
}
}
在頂部和底部添加一些經過的時間代碼。
long start = System.nanoTime();
…
System.out.println( "Elapsed: " Duration.ofNanos( System.nanoTime() - start ) );
在我的具有 8 個真正內核的M1 MacBook Pro上，在 Java 18 上，這大約需要 18 秒。
Count futures: 1000
Widget[id=56e594bf-75a6-4cf1-83fc-2b671873c534, whenCreated=2022-04-25T07:00:18.977719Z]
Widget[id=11373948-0689-467a-9ace-1e8d57f40f40, whenCreated=2022-04-25T07:00:18.977721Z]
…
Widget[id=d3b11574-6c11-41cc-9f26-c24ad53aa18c, whenCreated=2022-04-25T07:00:36.747058Z]
Widget[id=017ff453-da92-4296-992e-2c2a2ac44ed8, whenCreated=2022-04-25T07:00:36.748571Z]
Elapsed: PT17.906065583S
完整的示例代碼，方便您復制粘貼。
package work.basil.example.threading;
import java.time.Duration;
import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.*;
public class App
{
public static void main ( String[] args )
{
long start = System.nanoTime();
int threadPoolSize = 3; // Generally less than number of cores.
ExecutorService executorService = Executors.newFixedThreadPool( threadPoolSize );
record Widget( UUID id , Instant whenCreated )
{
}
Callable < Widget > makeWidgetTask = ( ) -> {
Thread.sleep( Duration.ofMillis( 50 ).toMillis() ); // Pretend that we have a long-running task.
Widget widget = new Widget( UUID.randomUUID() , Instant.now() );
return widget;
};
List < Callable < Widget > > tasks = Collections.nCopies( 1_000 , makeWidgetTask );
List < Future < Widget > > futures = null;
try
{
futures = executorService.invokeAll( tasks );
}
catch ( InterruptedException e )
{
throw new RuntimeException( e );
}
executorService.shutdown(); // Disable new tasks from being submitted.
try
{
if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
{
executorService.shutdownNow(); // Cancel currently executing tasks.
// Wait a while for tasks to respond to being cancelled.
if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
{ System.err.println( "Executor service did not terminate." ); }
}
}
catch ( InterruptedException ex )
{
executorService.shutdownNow(); // (Re-)Cancel if current thread also interrupted
Thread.currentThread().interrupt(); // Preserve interrupt status
}
System.out.println( "Count futures: " futures.size() );
for ( Future < Widget > future : futures )
{
if ( ! future.isDone() ) { System.out.println( "Oops! Task not done: " future.toString() ); }
else if ( future.isCancelled() ) { System.out.println( "Bummer. Task cancelled: " future.toString() ); }
else // Else task must have completed successfully.
{
try
{
Widget widget = future.get();
System.out.println( widget.toString() );
}
catch ( InterruptedException e )
{
throw new RuntimeException( e );
}
catch ( ExecutionException e )
{
throw new RuntimeException( e );
}
}
}
System.out.println( "Elapsed: " Duration.ofNanos( System.nanoTime() - start ) );
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/467462.html