Java-производитель/потребитель, обнаруживающий конец обработки

Question

Я готовлю приложение, в котором один производитель создает несколько миллионов задач, которые затем обрабатываются настраиваемым количеством потребителей. Связь от производителя к потребителю (вероятно) будет основана на очереди.

Из потока, который запускает продюсер/создает задачи, какой метод я могу использовать для ожидания выполнения всех задач? Я бы предпочел не возобновлять периодический опрос, чтобы проверить, пуста ли моя очередь задач. В любом случае пустая очередь задачи не является гарантией завершения последних задач. Эти задачи могут быть относительно длительными, поэтому вполне возможно, что очередь пуста, пока потребительские потоки все еще успешно обрабатывают.

Rgds, Маартен

Answer 1

Вы хотите знать, где каждые задачи завершается. У меня была бы очередная очередь завершенных отчетов задач. (Один объект/сообщение за задание) Когда этот счетчик достигает числа созданных вами задач, они все завершены. В этом отчете задачи также могут быть указаны любые ошибки и временная информация для задачи.

Answer 2

У вас может быть проверка каждого потребителя, чтобы убедиться, что очередь пуста, когда они деактивируются, и, если это так, пульсируйте конденсатор (или монитор, поскольку я считаю, что это то, что есть у Java), на котором ожидает основной поток ,

Имея потоки, проверяйте глобальную логическую переменную (помеченную как изменчивую), чтобы потоки знали, что они должны останавливаться.

Answer 3

Возможно, вам захочется взглянуть на пакет java.util.concurrent.

• ExecutorService
• Executors
• Future

Структура исполнитель уже предоставляет средства для выполнения задач с помощью ThreadPool. Абстракция Future позволяет дождаться завершения задач.

Включение обоих вместе позволяет легко координировать выполнение, развязывать задачи, действия (потоки) и результаты.

Пример:

ExecutorService executorService = Executors.newFixedThreadPool(16); 

    List<Callable<Void>> tasks = null; 
    //TODO: fill tasks; 

    //dispatch 
    List<Future<Void>> results = executorService.invokeAll(tasks); 

    //Wait until all tasks have completed 
    for(Future<Void> result: results){ 
     result.get(); 
    } 

Edit: Alternative Version с помощью CountDownLatch

ExecutorService executorService = Executors.newFixedThreadPool(16); 

    final CountDownLatch latch; 

    List<Callable<Void>> tasks = null; 
    //TODO: fill tasks; 

    latch = new CountDownLatch(tasks.size()); 

    //dispatch 
    executorService.invokeAll(tasks); 

    //Wait until all tasks have completed 
    latch.await(); 

А внутри ваших задач:

Callable<Void> task = new Callable<Void>() 
    { 

     @Override 
     public Void call() throws Exception 
     { 
      // TODO: do your stuff 

      latch.countDown(); //<---- important part 
      return null; 
     } 
    }; 

Answer 4

Вы можете использовать присоединиться() метод для каждого потока ..о том, что до тех пор, пока все нити не станут e ваш основной поток не закончится! И таким образом вы можете узнать, все ли потоки сделаны или нет!