Как можно заблокировать поток когда пул переполнен?

Question

У меня есть такой тестовый код:

    ExecutorService pool = Executors.newFixedThreadPool(1);

    System.out.println("Starting adding new messages to pool");
    for (int i = 0; i < 60; i++) {
        pool.submit(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
                System.out.println("Done!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    System.out.println("End of the program");
    Thread.sleep(10_000);

Если его запустить, то я увижу это:

Starting adding new messages to pool
End of the program
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!

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

Starting adding new messages to pool
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!

Как это можно сделать?

Answer 1

// завершаем работу ExecutorService (ExecutorService продолжит обрабатывать оставшийся список задач, но нельзя будет добавить новые).
pool.shutdown();
// блокируем текущий поток в ожидании завершения задач или таймаута, что произойдет раньше
pool.awaitTermination(600, TimeUnit.SECONDS);

Дополнение:

Если нужно блокировать цикл for, когда пул заполнен, то следует применить семафор:

final int threads = 1;
// сделаем семафор на threads разрешений
final Semaphore lock = new Semaphore(threads);
ExecutorService pool =  Executors.newFixedThreadPool(threads);

System.out.println("Starting adding new messages to pool");
for (int i = 0; i < 60; i++) {
    // запрашиваем разрешение, acquire() грубо говоря повышает счетчик в семафоре на +1, и если этот счетчик равен threads, то метод заблокирует поток, пока счетчик не станет меньше threads (а это произойдет при вызове release())
    lock.acquire();

    pool.submit(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(100);
            System.out.println(new Date() + " Done! ");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            finally {
                lock.release();
            }
        });


        // небольшой костылёк, который реализует ожидание последних threads (кол-во) задач 
        if (i == 60 - 1) {
            lock.acquire(threads);
        }
    }

    lock.release(threads);

    System.out.println("End of the program");
    Thread.sleep(10_000);

Answer 2

Cредствами Явы, без самописных велосипедов

UPD Без самописных велосипедов не обошлось, но размер велосипеда меньше, чем с семафорами.

// велосипед
private static boolean interruptibleInfiniteOffer(BlockingQueue<Runnable> q, Runnable r) {
    try {
        return q.offer(r, Long.MAX_VALUE, TimeUnit.NANOSECONDS); // infinite == ~292 years
    } catch (InterruptedException e) {
        return false;
    }
}

public static ThreadPoolExecutor getFixedSizePoolWithLimitedWaitingQueue(int nThreads, int maxItemsInTheQueue) {
    BlockingQueue<Runnable> queue = maxItemsInTheQueue == 0
            ? new SynchronousQueue<>() { public boolean offer(Runnable r) { return interruptibleInfiniteOffer(this, r);} }
            : new ArrayBlockingQueue<>(maxItemsInTheQueue) { public boolean offer(Runnable r) { return interruptibleInfiniteOffer(this, r);} };
    return new ThreadPoolExecutor(nThreads, nThreads, 0, TimeUnit.MILLISECONDS, queue);
}

public static void main(String[] args) throws InterruptedException {
    int nThreads = 4;
    int maxItemsInTheQueue = 1;
    ThreadPoolExecutor tpe = getFixedSizePoolWithLimitedWaitingQueue(nThreads, maxItemsInTheQueue);

    System.out.println("Starting adding new messages to pool");

    final Random random = new Random();
    for (int i = 0; i < 60; i++) {
        final int t = i;

        System.out.println("Adding task "+t+"...");
        tpe.submit(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(50 + random.nextInt(50));
                System.out.println("Task "+t+" is completed");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println("Task "+t+" is added");
    };

    tpe.shutdown();
    System.out.println("End of the program");
}

С помощью параметра maxItemsInTheQueue можно регулировать, сколько элементов может ожидать в очереди. Если 0, используем SynchronousQueue (задачи в очередь не попадают, продюсеры ждут освободившийся поток), если больше - то ArrayBlockingQueue (задачи попадают в очередь до заполнения, потом продюсеры начинают ждать освободившийся поток).

Дополнительно, можно offer(r, Long.MAX_VALUE, TimeUnit.NANOSECONDS) сделать настраиваемым так, что ожидание свободного слота в очереди будет не бесконечным, а с таймаутом.

БЛОГ НА HUSL

Как можно заблокировать поток когда пул переполнен?

Ответы (2 шт):