java - 如果需要处理的数据太多，如何让 ThreadPoolExecutor 命令等待？

Question

我正在从队列服务器获取数据，我需要处理它并发送确认。像这样:

while (true) {
    queueserver.get.data
    ThreadPoolExecutor //send data to thread
    queueserver.acknowledgement 

我不完全理解线程中发生了什么，但我认为这个程序获取数据，将其发送给线程，然后立即确认它。因此，即使我限制每个队列只能有 200 个未确认的项目，它也会尽可能快地接收它。当我在单个服务器上编写程序时，这很好，但如果我使用多个工作程序，那么这就会成为一个问题，因为线程队列中的项目数量不是它完成工作的反射(reflect)，而是它有多快可以从队列服务器获取项目。

如果线程队列充满工作，我能做些什么来让程序等待吗？

Answer 1

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

您可以使用具有限制的 BlockingQueue 而不是开放式队列:

BlockingQueue<Date> queue = new ArrayBlockingQueue<Date>(200);

就提交给 ExecutorService 的作业而言，不是使用使用无界队列的 Executors 创建的默认 ExecutorService，您可以创建自己的:

return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
              new ArrayBlockingQueue<Runnable>(200));

一旦队列填满，它将导致它拒绝任何提交的新任务。您需要设置一个提交到队列的 RejectedExecutionHandler。像这样的东西:

final BlockingQueue queue = new ArrayBlockingQueue<Runnable>(200);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(nThreads, nThreads,
           0L, TimeUnit.MILLISECONDS, queue);
// by default (unfortunately) the ThreadPoolExecutor will throw an exception
// when you submit the 201st job, 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
      executor.getQueue().put(r);
      // check afterwards and throw if pool shutdown
      if (executor.isShutdown()) {
         throw new RejectedExecutionException(
              "Task " + r + " rejected from " + e);
      }
   }
});

我认为 Java 没有 ThreadPoolExecutor.CallerBlocksPolicy 是一个重大失误。