是否可以限制提交到线程池的任务数量？

Question

例如，我有一个有100个线程的线程池的场景。

有10个作业，每个作业都可以向线程池发送1..n个任务。

如果我只是直接提交，他们将工作竞争线程在池中。

我有什么办法可以说：

作业1可以同时向线程池提交最多5个任务，并且在发送下一个任务之前必须等待其中一个任务完成。

我知道，如果我对每个作业都有单独的线程池，我就可以做到这一点。但是，这些作业是动态出现和消失的传入请求。以这种方式动态创建线程池可能不太好。

我可以使用一个大线程池来实现上面的目标吗？

Answer 1

您可以创建自己的ExecutorService，如下所示：

class LimitingExecutorService implements ExecutorService {
  private final ExecutorService delegate;
  private final Semaphore semaphore;

  LimitingExecutorService(ExecutorService delegate, int limit) {
    this.delegate = delegate;
    this.semaphore = new Semaphore(limit);
  }

现在，您可以实现将调用委托给delegate的方法，但是可以检查是否可以获取信号量，例如：

  public Future<?> submit(Runnable task) {
    // Or you could block.
    if (!semaphore.tryAcquire()) {
      throw new RejectedExecutionException(...); // Indicate that the task couldn't be submitted.
    }

    // Wrap task with in another runnable() that releases the semaphore (whether or not it succeeds).
    try {
      return delegate.submit(() -> {
        try {
          task.run();
        } finally {
          semaphore.release();
        }
      });
    } catch (RejectedExecutionException e) {
      semaphore.release();
      throw e;
    }
  }

等其他方法。

您将需要关注像invokeAll这样的方法，以决定行为应该是什么:应该尽可能多地调用它们，或者只有在所有任务都能在此时被调度时，调用才能成功。

现在，您的每个作业都可以有自己的LimitingExecutorService实例，并且它们只能提交信号量允许的多个作业。