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

Question

我有下面的代码片段，它运行良好。但问题是它立即创建并在执行程序队列中放置了 2000 多个任务。

我需要检查执行程序队列中的任务是否已完成，然后才给它更多的任务。它不必非常精确，即如果队列还剩 <10 个任务，则再添加 50 个。

所以executor任务队列没有那么多pending任务，这也会让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

I have the below code snippet, which runs fine. But the problem is it creates and put over 2000 tasks on the executor queue right off the bat.

实现此目的的一种方法是创建您自己的具有有限作业队列的 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 中的核心线程数和最大线程数设置为相同的数字。不幸的是，默认情况下，执行器启动核心线程，然后填充队列，然后才分配额外的线程直到达到最大值。这完全违反直觉。