Блог

У меня есть сценарий, в котором у меня есть пул потоков, например, из 100 потоков.

Существует 10 заданий, каждое из которых может отправлять 1..n задач в пул потоков.

Если я просто отправлю их напрямую, они будут конкурировать за потоки в пуле.

Есть ли какой-нибудь способ, которым я могу сказать что-то вроде:

Задание 1 может одновременно отправлять в пул потоков не более 5 задач, и перед отправкой следующей из них необходимо дождаться завершения одной из них.

Я знаю, что смогу это сделать, если у меня будут отдельные пулы потоков для каждого задания. Но эти задания-это входящие запросы, которые появляются и исчезают на лету. Возможно, не стоит динамически создавать пулы потоков таким образом.

Могу ли я достичь вышеуказанного, используя один большой пул потоков?

Вы можете создать свой собственный 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 , и они могут отправлять столько заданий , сколько позволяют доступные разрешения семафора.