исполнители с общей очередью

Question

У меня есть один производитель задач и несколько аппаратных ресурсов для их выполнения. Итак, я пытаюсь создать исполнителей с общей очередью

BlockingQueue<Runnable> queue;
ExecutorService executor1 = new ThreadPoolExecutor(poolSize1, poolSize1, 0L, TimeUnit.MILLISECONDS, queue, threadFactory);
ExecutorService executor2 = new ThreadPoolExecutor(poolSize2, poolSize2, 0L, TimeUnit.MILLISECONDS, queue, threadFactory);
ExecutorService executor3 = new ThreadPoolExecutor(poolSize3, poolSize3, 0L, TimeUnit.MILLISECONDS, queue, threadFactory);

и (просто попробуйте) добавить задачу в queue.put (задача) вместо конкретной executorN.execute (задача) Но execurer не создает поток, пока execute (task) call.

Итак, мне нужна реализация Executor? который запускает задачу на любом свободном потоке суб-исполнителя. Может быть, вы знаете решение / библиотеку, которую можно использовать?

Это не балансировщик. Мне все равно, где задача начнется. Пусть он будет стоять в очереди, пока его не обработает свободный обработчик

Answer 1

Я бы предпочел, чтобы вы реализовали это вручную:

    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(500);

    IntStream.range(0, 10).forEach(i -> {
      Thread t = new Thread(() -> {
        while (true) {
          try {
            queue.take().run();
          } catch (InterruptedException e) {
            break;
          }
        }
      });
      t.setDaemon(true);
      t.setName("worker-"+i);
      t.start();
    });

    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));

    Thread.sleep(1000);

    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));
    queue.add(() -> System.out.println(Thread.currentThread().getName()));

    Thread.sleep(1000);

, и результат будет:

worker-0
worker-1
worker-2
worker-3
worker-4
worker-5
worker-6
worker-7
worker-8
worker-9
worker-0
worker-1
worker-2

Потоки PS будут ждать нового элемента в очереди.