如何不压倒 java executorservice 任务队列？

Question

我有下面的代码 sn-p，它运行良好。但问题是它立即创建并将超过 2000 个任务放在执行程序队列中。

我需要检查已经在执行器队列中的任务是否完成，然后再给它更多任务。不一定要准确，即如果队列剩余

所以executor任务队列没有那么多待处理的任务，这样也可以让shutdown()及时工作，否则即使被调用，executor还是会先尝试完成队列中的所有2000个任务。

完成此任务的最佳方法是什么？谢谢

executor = Executors.newFixedThreadPool(numThreads);

while(some_condition==true)
{
    //if(executor < 10 tasks pending)  <---- how do i do this?
    //{                             
        for(int k=0;k<20;k++)
        {  
            Runnable worker = new MyRunnable();
            executor.execute(worker);
        }
    //}
    //else 
    //{
    //      wait(3000);
    //}
} 

---

使用信号量更新：

private final Semaphore semaphore = new Semaphore(10)
executor = new ThreadPoolExecutorWithSemaphoreFromJohnExample();

while(some_condition==true)
{

        Runnable worker = new MyRunnable();
        //So at this point if semaphore is full, then while loop would PAUSE(??) until
        //semaphore frees up again.
          executor.execute(worker);   
} 

Answer 1

我有下面的代码 sn-p，它运行良好。但问题是它会立即创建 2000 多个任务并将其放入执行程序队列中。

一种方法是使用有限的作业队列创建您自己的ThreadPoolExecutor，并在其上设置自定义RejectedExecutionHandler。这使您可以细粒度地控制要排队的作业数量。

您需要自定义处理程序，因为默认情况下，如果队列已满，ThreadPoolExecutor.submit(...) 将抛出 RejectedExecutionException。使用下面的自定义处理程序，当它被队列拒绝时，拒绝处理程序只是将其放回原处，阻塞直到队列有空间。所以不会有任何工作被拒绝/放弃。

这里大致介绍了如何启动自己的线程池并设置自己的拒绝处理程序。

// you can tune the blocking queue size which is the number of jobs to queue
// when the NUM_THREADS are all working
final BlockingQueue<MyRunnable> queue =
    new ArrayBlockingQueue<MyRunnable>(NUM_JOBS_TO_QUEUE);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(NUM_THREADS, NUM_THREADS,
       0L, TimeUnit.MILLISECONDS, queue);
// by default (unfortunately) the ThreadPoolExecutor will throw an exception
// when you submit the job that fills the queue, to have it block you do:
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      // this will block if the queue is full as opposed to throwing
      executor.getQueue().put(r);
   }
});
...
// now submit all of your jobs and it will block if the queue is full
for(int k = 0; k < 20000000; k++) {  
   Runnable worker = new MyRunnable();
   threadPool.execute(worker);
}

有关阻塞线程池的更多详细信息，请参见我的答案：

How can I make ThreadPoolExecutor command wait if there's too much data it needs to work on?

您还可以使用ThreadPoolExecutor.CallerRunsPolicy，这将导致将作业提交到线程池的调用者执行作业。但是我不喜欢这个解决方案，因为它会阻塞调用者，直到作业完成，这可能会使其他工作线程饿死。此外，如果有多个提交者，仍然可能导致运行作业的线程过多。

最后，请注意我将ThreadPoolExecutor 中的核心和最大线程数设置为相同的数字。不幸的是，默认情况下，执行程序会启动核心线程，然后填充队列，然后它才会分配额外的线程，直到达到最大值。这完全违反直觉。

Answer 2

您可以使用简单的信号量。提交后获取新的许可证，完成后释放许可证以允许其他等待提交的人。

private final Semaphore semaphore = new Semaphore(10);//or however you want max queued at any given moment
ThreadPoolExecutor tp= new ThreadPoolExecutor(...){
      public void execute(Runnable r){
          semaphore.acquire();
          super.execute(r);
      }    
      public void afterExecute(Runnable r, Thread t){
         semaphore.release();  
         super.afterExecute(r,t);
      }
};

因此，如果没有更多可用的许可，提交线程将被暂停。

Answer 3

我通常通过对任务对象使用对象“池队列”来限制此类系统 - 一个在启动时充满 X 个任务的 BlockingQueue。任何想要向线程提交任务的东西都必须从池队列中获取一个，将数据加载到其中然后提交。

当任务完成并导致它被处理后，它会被推回池队列以供重复使用。

如果池为空，提交线程会阻塞在池队列中，直到返回一些任务。

这本质上是@John Vint 建议的一种信号量控制形式，但还有一些进一步的优点 - 例如，无需持续创建/GC 可运行对象。我喜欢将 PooolQueue.size 转储到计时器上的 GUI 状态栏，这样我就可以看到系统有多“忙”，（并且还可以快速检测任何对象泄漏：）

Answer 4

您最好设置拒绝策略，因为您不想让线程池不堪重负，我发现在不让自己复杂化的情况下完成此操作的最佳方法是执行以下操作：

final ThreadPoolExecutor executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(THREADS_COUNT);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

会发生什么，一旦所有线程都忙了，调用者的线程就会执行任务。这是对此类政策的参考CallerRunsPolicy JavaDoc