怎么不压倒java executorservice任务队列？

Question

我有下面的代码片段,运行正常.但问题是它立即创建并在执行程序队列上放置了2000多个任务.

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

因此执行程序任务队列没有那么多挂起任务,这也将允许shutdown()及时工作,否则即使被调用,执行程序仍将首先尝试完成其队列中的所有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

我有下面的代码片段,运行正常.但问题是它立即创建并在执行程序队列上放置了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);
}

有关阻塞线程池的更多详细信息,请参阅我的答案:

如果需要处理太多数据,我如何让ThreadPoolExecutor命令等待？

您还可以使用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);
      }
};

因此,如果没有更多可用许可,则提交线程将被暂停.