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 не рассматриваете ? 22 дек 2016 в 10:54
  • Не понимаю как он мне может помочь.... Дополнил вопрос.
    – faoxis
    22 дек 2016 в 11:01
  • проще складывать в пул скачивание каждой ссылки индивидуально, и awaitTermination, а для ограничения числа потоков использовать fixedThreadPool.
    – zRrr
    22 дек 2016 в 13:25
  • @zRrr не могли бы вы привести пример ? Пока не очень понятно как это выглядеть.
    – faoxis
    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
    22 дек 2016 в 13:03
  • @faoxis вы не можете ждать на основе AtomicX без спинлупа, а такое, в общем, нигде не приветствуется. И волатайл здесь не нужен, атомик и так атомик.
    – etki
    22 дек 2016 в 13:10
  • Да, действительно, volatile лишний. Я не понимаю, что мешает в вашем примере произойти следующему: в конце остаются, скажем, 2 потока. Один забирает последний элемент, второй освобождает ожидание (synchronizer.notify()). Дальше все нити отпускаются и программа завершается не смотря на то, что первый поток все еще обрабатывает сообщение из очереди.
    – faoxis
    22 дек 2016 в 13:26
  • @faoxis да, это факап. лучше с CountDownLatch.
    – etki
    22 дек 2016 в 14:00
  • а куда я прикручу CountDownLatch ? Если сделаю статикой, то чем тогда такой подход отличается от того, что я предлажил выше. А если не статика, то как смежные объекты будут знать о друг друге ?
    – faoxis
    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
    22 дек 2016 в 14:07
  • @faoxis они не умрут, пока у треда есть работа он будет активен. И метод getActiveCount будет возвращать отличное от нуля число 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
    22 дек 2016 в 17:34

Ваш ответ

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

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