ThreadPoolExecutor 未执行所有任务

Question

我有一个 ThreadPoolExecutor -corePoolSize=5, maxPoolSize=10 queueSize = 10,keepAlive=1000 秒。我正在执行100 Runnable 任务。实际执行的任务数量是变化的，并不是所有的都被执行。 RejectionHandler 也没有报告任何内容。我相信我对 ThreadPoolExecutor 的理解在某种程度上是错误的。有谁能够帮我？我怎样才能让所有的任务都被执行？

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestThreadPoolExecutor {
    public static void main(String[] args) {
     ThreadFactory threadFactory =  Executors.defaultThreadFactory();
    ArrayBlockingQueue<Runnable> arrayBlockingQueue = new ArrayBlockingQueue<Runnable>(10);

    ThreadPoolExecutor threadPoolExecutor =  new ThreadPoolExecutor(5, 10,1000, TimeUnit.SECONDS, arrayBlockingQueue, threadFactory, new RejectedExecutionHandlerImpl());
    MonitorThread monitor = new MonitorThread(threadPoolExecutor, 3);
    Thread monitorThread = new Thread(monitor);
    monitorThread.start();
    for (int i = 0; i < 100; i++) {
        threadPoolExecutor.execute(new DummyRunnableTask(i));
    }
    //threadPoolExecutor.shutdown();
    //monitor.shutDown();
}
}

class DummyRunnableTask implements Runnable {

  private int i;

    public DummyRunnableTask(int i) {
    super();
    this.i = i;
  }

    @Override
    public void run() {
    /*try {
        Thread.sleep(1);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }*/
    System.out.println("Thread Name:=" + Thread.currentThread().getName()+ " is working for id=" + i);
}

}

class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {

    @Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    System.out.println();
}

}

class MonitorThread implements Runnable{
    private ThreadPoolExecutor executor;
    private int seconds;
    private Boolean run = true;
    public MonitorThread(ThreadPoolExecutor executor, int seconds) {
    super();
    this.executor = executor;
    this.seconds = seconds;
  }

    public void shutDown() {
    this.run = false;
  }

    @Override
    public void run() {
    while (run) {
        System.out.println(
                String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
                    this.executor.getPoolSize(),
                    this.executor.getCorePoolSize(),
                    this.executor.getActiveCount(),
                    this.executor.getCompletedTaskCount(),
                    this.executor.getTaskCount(),
                    this.executor.isShutdown(),
                    this.executor.isTerminated()));
            try {
                Thread.sleep(seconds*1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    }

  }

 }

样本输出：

...............
Thread Name:=pool-1-thread-7 is working for id=97
Thread Name:=pool-1-thread-2 is working for id=95
Thread Name:=pool-1-thread-10 is working for id=94
Thread Name:=pool-1-thread-4 is working for id=93
[monitor] [0/5] Active: 0, Completed: 0, Task: 1, isShutdown: false, isTerminated: false
[monitor] [10/5] Active: 0, Completed: 88, Task: 88, isShutdown: false, isTerminated: false
[monitor] [10/5] Active: 0, Completed: 88, Task: 88, isShutdown: false, isTerminated: false

Answer 1

我很欣赏 @Speise 提供的答案，但只是为了添加一些清晰明了的细节；我会引用 oracle 文档中提到的声明：

如果运行的线程少于 corePoolSize，则 Executor 总是更喜欢添加新线程而不是排队。

如果 corePoolSize 或更多线程正在运行，Executor 总是更喜欢排队请求而不是添加新线程。

如果一个请求不能入队，就会创建一个新线程，除非这会超过 maximumPoolSize，在这种情况下，该任务将被拒绝。

文档非常清晰，并指出为了避免拒绝任务，您应该增加 maxPoolSize。 而不是增加 Queue 大小，因为可能不会出现这种情况（如由您在 cmets 中）当您事先知道将要提交多少任务时（因此您可以调整 maxPoolSize）。我还建议您在这种类型的场景中使用UnboundedQueues(LinkedBlockingQueue)（但是；对于这种情况可能有一个快速解决方案，您也可以增加将请求推送到队列的时间）。

同时增加队列大小也会影响性能，所以为什么不将它（相应地创建线程）留给 JVM。（因为ThreadPool 根据maxPoolSize 采取行动）

Answer 2

假设 100 个线程不能被处理为 maxPoolSize=10 和 queueSize = 10，这意味着您可以在最坏的情况下只放入 20 个线程。最佳情况可以根据每个线程内作业的性能和复杂性而改变。尝试将您的 queueSize 增加到 90。所以肯定有 90 个会等待，其他 10 个会继续工作。 你可以在这里找到最好的解释link:

如果请求无法排队，则创建一个新线程，除非这 将超过 maximumPoolSize，在这种情况下，任务将 被拒绝。

Answer 3

TestThreadPoolExecutor 参数：

corePoolSize - 保留在池中的线​​程数，即使它们处于空闲状态。

ma​​ximumPoolSize - 池中允许的最大线程数。

keepAliveTime - 当线程数大于核心时，这是多余的空闲线程在终止前等待新任务的最长时间。

unit - keepAliveTime 参数的时间单位。

workQueue - 用于在执行任务之前保存任务的队列。此队列将仅保存由 execute 方法提交的 Runnable 任务。

handler - 由于达到线程边界和队列容量而阻塞执行时使用的处理程序。

您的场景： corePoolSize = 5，maximumPoolSize = 10，工作队列 = 10

你在 for 循环中推送 100 个任务。

首先，您永远不能获得超过 10 个工作线程。

假设你所有的工作线程都忙，那么最多可以容纳 10 个工作队列。在那之后，任何提交的任务都将被拒绝，你可以在拒绝处理程序中捕获该拒绝。 因此增加工作队列将有助于解决问题。 （但不推荐）

你可以尝试什么：

在推送您的工作时使用一些时间间隔，而不是使用 for 循环，这实际上是在尝试在同一实例推送工作。

将您的 maximumPoolSize 设置为核心数 X 2（或 4）。

当您的作品被拒绝时，您可以尝试延迟一段时间后重新提交作品

PS：这是生产者-消费者问题。它总是要求消费者比生产者快。有一段时间，生产者变得更快，我们看到工作量激增。为了处理负载峰值，设计了队列，但即使队列溢出，拒绝也会被发送回生产者，生产者有责任通过重新提交或丢弃来优雅地处理它。

Answer 4

感谢@Mandy8055、@Ritesh、@Speise

我找到了两种方法，如果无法预测 queueSize，可以优雅地处理此问题。

1. 编写一个拒绝处理程序，将任务提交回队列。但是话又说回来，正如@Ritesh 指出的那样，生产者可能比消费者更快，我们最终可能会得到一个 StackOverFlow。 无论如何，这就是我所做的：

   class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
   
   @Override
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
   DummyRunnableTask task =  (DummyRunnableTask)r;
   
   System.out.println("Task id "+task.getI()+" got rejected, resubmitting");
   try {
       Thread.sleep(100);
   } catch (InterruptedException e) {
       e.printStackTrace();
   }
   executor.execute(r);
   }
   }
   

   现在，这并不是处理它的好方法，因为您只是不知道该任务是否需要 100 毫秒才能完成。也许我们可以创建一个被拒绝的任务列表，然后重新将它们全部重新提交。

2. 另一种方法是阻塞ThreadPoolExecutor的提交/执行方法，一旦达到一定的限制（maxPoolSize+queueSize），让它等待直到有空间。我们通过在 ThreadPoolExecutor 的自定义实现中使用 Bounded Semaphore 来做到这一点

   class CustomExecutor { 
   
   private final ThreadPoolExecutor executor;
   private final Semaphore semaphore;
   
   public CustomExecutor(int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit unit,int queueSize, ThreadFactory factory) {
   BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(queueSize);
   this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, queue, factory);
   this.semaphore = new Semaphore(queueSize + maxPoolSize);
   }
   public void shutDown(){
   this.executor.shutdown();
   }
   
   public ThreadPoolExecutor getExecutor() {
   return executor;
   }
   
   
   private void exec(final Runnable command) throws InterruptedException {
   semaphore.acquire();
   try {
       executor.execute(new Runnable() {
           @Override
           public void run() {
               try {
                   command.run();
               } finally {
                   semaphore.release();
               }
           }
       });
   } catch (RejectedExecutionException e) {
       semaphore.release();
       throw e;
   }
   }
   
   public void execute (Runnable command) throws InterruptedException {
   exec(command);
     }
   }
   

如果您能想到其他方法，请告诉我们。