0

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

    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!

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

2
  • продьюсер пула потоков блокировался это имеется ввиду, что submit должен блокироваться, если нет свободных потоков? 3 мар 2021 в 23:40
  • @RomanKonoval да 4 мар 2021 в 5:16

2 ответа 2

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);
4
  • Но мне не надо завершать работу ExecutorService, мне надо просто блокировать поставщика данных в поток. 4 мар 2021 в 5:16
  • Обновил, посмотрите
    – lampa
    4 мар 2021 в 6:37
  • 1
    а я уже начал делать через SynchronousQueue. Выше решение более простое, спасибо. :) 4 мар 2021 в 7:22
  • @ВасилийНикпуп Это каким боком семафоры проще-то? Ты правильно стал копать в сторону SynchonousQueue, но не докопал. См. мой ответ 6 мар 2021 в 21:39
0

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) сделать настраиваемым так, что ожидание свободного слота в очереди будет не бесконечным, а с таймаутом.

2
  • Что-то не работает onlinegdb.com/SJJmJVGXd
    – lampa
    7 мар 2021 в 10:47
  • Да, забыл за давностью лет, что ThreadPool использует offer, который не ждёт. Поправил ответ. 7 мар 2021 в 14:13

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.