3

Есть вот такой вот участок кода:

    BlockingQueue<String> queue = new ArrayBlockingQueue(links.size());
    for (String link : links) {
        queue.add(link);
    }

    ExecutorService executorService = Executors.newCachedThreadPool();
    int threadsNumber = Integer.valueOf(arguments.get("-n"));

    for (int i = 0; i < threadsNumber; i++) {
        executorService.submit(new Consumer(queue, arguments.get("-o"), Integer.valueOf(arguments.get("-l"))));
    }

    while (!queue.isEmpty()) {
        Thread.sleep(1000);
    }

    executorService.shutdownNow();
    System.out.println("Все файлы были успешно скачены");

Последние строчки означают, что я собираюсь закрыть пул с потоками когда все сообщения из очереди будут прочитаны. Все отлично, но возникает проблема в момент, когда один из потоков забирает последнее сообщение из очереди. Условие !queue.isEmpty() перестает выполняться и программа завершается до того как это сообщение обработалось.

Как мне сделать так, чтобы программа завершалась только после обработки всех сообщений в очереде ?

Метод run в Consumer

@Override
public void run() {
    try {
        while (true) {
            if (!queue.isEmpty()) {
                String stringURL = queue.take();
                String inputFileName = Paths.get(stringURL).getFileName().toString();
                String outputFileName = new File(new File(storageFilesDirectory), inputFileName).toString();

                Downloader.download(stringURL, outputFileName, speedLimit);
                System.out.println(String.format("Файл %s был успешно скачен", inputFileName));
            }
            Thread.sleep(500);
        }
    } catch (InterruptedException e) {

    } catch (FileCouldNotBeDownloaded e) {
        System.out.println(String.format("Произошла ошибка скачивания файла."));
    }
}

В пуле у меня всего два потока, а сообщений в очереди может быть неограничено.

4
  • 3
    executorService.awaitTermination не рассматриваете ? Commented 22 дек. 2016 в 10:54
  • Не понимаю как он мне может помочь.... Дополнил вопрос.
    – faoxis
    Commented 22 дек. 2016 в 11:01
  • проще складывать в пул скачивание каждой ссылки индивидуально, и awaitTermination, а для ограничения числа потоков использовать fixedThreadPool.
    – zRrr
    Commented 22 дек. 2016 в 13:25
  • @zRrr не могли бы вы привести пример ? Пока не очень понятно как это выглядеть.
    – faoxis
    Commented 22 дек. 2016 в 13:42

3 ответа 3

3

Обычно пулы потоков завершаются только при завершении самого приложения: автор приложения знает, что у него есть некоторая фоновая обработка задач, пул слушателей базы данных, поэтому можно их один раз создать и не беспокоиться о них - в худшем случае потоки просто будут простаивать, занимая какое-то небольшое количество оперативки. Если вы оставите сам пул потоков живым до полного завершения приложения, на работе самого приложения это не скажется, а вот работать с ним будет легче, потому что не придется иметь в виду его жизненный цикл.

В вашем случае вы ищете не момент завершения пула потоков, а некоторый синхронизатор, который вам скажет о том, что все потребители отработали. Проще всего завести CountDownLatch, инициализировав его с помощью размера очереди, и снимая по единице при каждой обработке сущности из очереди - в этом случае вы можете с помощью него дождаться конца обработки. Можно обойтись и более низкоуровневыми примитивами (никогда с ними не работал, поэтому могу ошибиться):

public class Consumer implements Runnable {
    private final BlockingQueue<String> queue;
    private final Object synchronizer;

    public Consumer(BlockingQueue<String> queue, Object synchronizer) {
        this.queue = queue;
        this.synchronizer = synchronizer;
    }

    public void run() {
        String value;
        while ((value = queue.poll(0, TimeUnit.Milliseconds)) !== null) {
            // обработка
        }
        // в эту область код попадет только тогда, когда в очереди кончатся элементы
        synchronizer.notify();
    }
}

// код главного потока

for (Consumer consumer : consumers) {
    executor.submit(consumer);
}

synchronizer.wait();
5
  • Тогда уж можно еще проще: public static volatile AtomicInteger handledCounter в Consumer. :)
    – faoxis
    Commented 22 дек. 2016 в 13:03
  • @faoxis вы не можете ждать на основе AtomicX без спинлупа, а такое, в общем, нигде не приветствуется. И волатайл здесь не нужен, атомик и так атомик.
    – etki
    Commented 22 дек. 2016 в 13:10
  • Да, действительно, volatile лишний. Я не понимаю, что мешает в вашем примере произойти следующему: в конце остаются, скажем, 2 потока. Один забирает последний элемент, второй освобождает ожидание (synchronizer.notify()). Дальше все нити отпускаются и программа завершается не смотря на то, что первый поток все еще обрабатывает сообщение из очереди.
    – faoxis
    Commented 22 дек. 2016 в 13:26
  • @faoxis да, это факап. лучше с CountDownLatch.
    – etki
    Commented 22 дек. 2016 в 14:00
  • а куда я прикручу CountDownLatch ? Если сделаю статикой, то чем тогда такой подход отличается от того, что я предлажил выше. А если не статика, то как смежные объекты будут знать о друг друге ?
    – faoxis
    Commented 22 дек. 2016 в 14:10
3

Используйте java.util.concurrent.ThreadPoolExecutor, он гибче в использовании.

Пример для вашей задачи:

ThreadPoolExecutor poolExecutor = 
            new ThreadPoolExecutor(4, 4, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

for (int i = 0; i < 1_000_000; i++) 
     poolExecutor.execute(()->{
        //задача выполняемая в другом потоке
     });

while (poolExecutor.getActiveCount() != 0) 
    LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));

poolExecutor.shutdownNow();
2
  • У меня есть k потоков, которые должны скачать n файлов. n может быть намного больше k. Мне не надо, чтобы обработчик умирал после работы.
    – faoxis
    Commented 22 дек. 2016 в 14:07
  • @faoxis они не умрут, пока у треда есть работа он будет активен. И метод getActiveCount будет возвращать отличное от нуля число
    – Artem
    Commented 22 дек. 2016 в 15:35
2

Если нужно выполнить N заранее известных задач в K потоков, то проще использовать Executor с фиксированным числом потоков и скормить ему N Runnable выполняющих одну задачу:

final int NUMBER_OF_WORKERS = 2;
List<String> urls = IntStream.range( 0, 20 ).mapToObj( String::valueOf ).collect( Collectors.toList() );

ExecutorService executor = Executors.newFixedThreadPool( NUMBER_OF_WORKERS );
for ( String url : urls ) {
    executor.submit( () -> { // new SingleUrlDownloadTask( url ) etc...
        System.out.printf( "Worker [%s] downloading url: %s%n", Thread.currentThread().getName(), url );
        try {
            Thread.sleep( 300 + ThreadLocalRandom.current().nextInt(400) );
        } catch (Exception e) {}
        System.out.printf( "Worker [%s] completed url: %s%n", Thread.currentThread().getName(), url );
    });
}

System.out.println( "All tasks queued." );

executor.shutdown(); // пул перестает принимать новые задачи,
                     // уже поставленные в очередь задачи будут выполнены рано или поздно
executor.awaitTermination( Long.MAX_VALUE, TimeUnit.NANOSECONDS );

System.out.println( "All tasks done.");
1
  • Ничего себе примерчик. :) Я всего лишь хочу устроиться Java Junior'ом. :) Долго я его ковырял, но прокачался хорошо. Спасибо!
    – faoxis
    Commented 22 дек. 2016 в 17:34

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

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