為什么我不能在Java中第二次向執行緒池添加任務？-有解無憂

我創建了一個執行緒池來處理任務，處理完任務后發現無法添加和啟動其他任務？如何解決？如果我更改執行executor = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("timeOutThread")); 器，它會運行正常。但是如果任務因為超時而被取消，這會導致記憶體泄漏嗎？

ExecutorService executor =   new ThreadPoolExecutor(3,
                    3, 0L,
                    TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1),
                    new NamedThreadFactory(
                            "timeOutThread")); 
    List<Callable<String>> callableList = new ArrayList<>();
    IntStream.range(0, 3).forEach(index -> {
        callableList.add(() -> request(index));
    });
    List<Future<String>> futureList = executor.invokeAll(callableList, 1, TimeUnit.SECONDS); 
        for (int i = 0; i < futureList.size(); i  ) {
            Future<String> future = futureList.get(i);
            try {
                list.add(future.get());
            } catch (CancellationException e) {
                log.info("timeOut task：{}", i);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
          Thread.sleep(1000);
    callableList.clear();
    IntStream.range(0, 3).forEach(index -> {
        callableList.add(() -> request(index));
    });
      long start1 = System.currentTimeMillis();
     // Task java.util.concurrent.FutureTask@5fdcaa40 rejected from java.util.concurrent.ThreadPoolExecutor@6dc17b83   
    List<Future<String>> futureList   = executor.invokeAll(callableList, 1, TimeUnit.SECONDS); 
    for (int i = 0; i < futureList.size(); i  ) {
        Future<String> future = futureList.get(i);
        try {
            list.add(future.get());
        } catch (CancellationException e) {
            log.info("timeOut Task：{}", i);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

  public String request() throws InterruptedException { 
    TimeUnit.MILLISECONDS.sleep(200000);
    return "A";
  }

uj5u.com熱心網友回復：

我可以使用以下簡化代碼重現您的錯誤：

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Main {

  public static void main(String[] args) throws InterruptedException {
    var pool = new ThreadPoolExecutor(
        3, 3, 0L, TimeUnit.NANOSECONDS, new LinkedBlockingQueue<>(1));
    try {
      System.out.println("Executing first batch of tasks...");
      submitTasks(pool);

      System.out.println("Executing second batch of tasks...");
      submitTasks(pool);
    } finally {
      pool.shutdown();
    }
  }

  private static void submitTasks(ExecutorService executor) throws InterruptedException {
    var tasks = new ArrayList<Callable<Void>>(3);
    for (int i = 0; i < 3; i  ) {
      tasks.add(() -> {
        Thread.sleep(2_000L);
        return null;
      });
    }
    executor.invokeAll(tasks);
  }
}

這給出了這個輸出：

Executing first batch of tasks...
Executing second batch of tasks...
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@87aac27[Not completed, task = Main$$Lambda$1/0x0000000800c009f0@816f27d] rejected from java.util.concurrent.ThreadPoolExecutor@3e3abc88[Running, pool size = 3, active threads = 0, queued tasks = 1, completed tasks = 3]
        at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2070)
        at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833)
        at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365)
        at java.base/java.util.concurrent.AbstractExecutorService.invokeAll(AbstractExecutorService.java:247)
        at Main.submitTasks(Main.java:32)
        at Main.main(Main.java:18)

問題是由于佇列太小造成的。創建的LinkedBlockingQueue只有一個容量，但一次將三個任務提交到池中。那么，問題就變成了，為什么它只在第二次呼叫時失敗invokeAll？

原因與如何ThreadPoolExecutor實施有關。首次創建實體時，不會啟動任何核心執行緒。它們在提交任務時懶惰地啟動。當一個任務的提交導致一個執行緒被啟動時，該任務會立即交給該執行緒。佇列被繞過。因此，當invokeAll第一次呼叫時，三個核心執行緒中的每一個都被啟動，并且沒有任何任務進入佇列。

但是第二次invokeAll呼叫，核心執行緒已經啟動了。由于提交任務不會導致創建執行緒，因此將任務放入佇列中。但是佇列太小，導致RejectedExecutionException. 如果您想知道為什么盡管保持活動時間設定為零，但核心執行緒仍然處于活動狀態，那是因為默認情況下不允許核心執行緒因超時而死亡（您必須顯式配置池以允許這樣做）。

通過稍微修改代碼，您可以看到這個 lazily-started-core-threads 是問題的原因。只需添加：

pool.prestartAllCoreThreads();

就在創建池之后，導致第一次呼叫invokeAll現在失敗并帶有RejectedExecutionException.

此外，如果您將佇列的容量從 1 更改為 3，則將RejectedExecutionException不再發生。

以下是一些相關檔案：

AnyBlockingQueue可用于傳輸和保存提交的任務。此佇列的使用與池大小互動：

corePoolSize如果運行的執行緒數少于此，則Executor總是更喜歡添加新執行緒而不是排隊。

如果corePoolSize或更多執行緒正在運行，則Executor總是更喜歡排隊請求而不是添加新執行緒。

如果請求無法排隊，則會創建一個新執行緒，除非超過maximumPoolSize，在這種情況下，該任務將被拒絕。

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/496647.html

標籤：爪哇多线程线程池线程池执行器

上一篇：從多個執行緒訪問同一個物件實體

下一篇：多執行緒：通過條件變數進行通信時，哪些變數需要互斥鎖保護？