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

Question

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

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

Rgds, Maarten

Answer 1

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

• ExecutorService
• Исполнители
• Future

Среда executor уже предоставляет средства для выполнения задач через пул потоков.Абстракция 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();
    }

Редактировать: Альтернативная версияиспользуя 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 2

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

Проверка потоками глобальной логической переменной (помеченной как volatile) - это способ сообщить потокам, что они должны остановиться.

Answer 3

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

Answer 4

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