ExecutorService никогда не останавливается без исключений

Question

Я принял стратегию параллелизма с этого поста . Однако мой выглядит так:

ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_CREATE_KNOWLEDGE_THREADS);
List<Callable<Collection<Triple>>> todo = new ArrayList<Callable<Collection<Triple>>>(this.patternMappingList.size());
for (PatternMapping mapping : this.patternMappingList ) {
    todo.add(new CreateKnowledgeCallable(mapping, i++));
}
try {

    List<Future<Collection<Triple>>> answers = executorService.invokeAll(todo);
    for (Future<Collection<Triple>> future : answers) {

        Collection<Triple> triples = future.get();
        this.writeNTriplesFile(triples);
    }    
}
catch (InterruptedException e) { ... }
catch (ExecutionException e) { ... }

executorService.shutdown();
executorService.shutdownNow();

Но ExecutorService никогда не выключается. Я попытался отладить, сколько из CreateKnowledgeCallable завершено, но это число, кажется, меняется (после того, как новые потоки / вызовы не выполняются, но служба продолжает работать). Я уверен, что вошли и распечатали все возможные исключения, но я не вижу, чтобы это произошло. Также кажется, что через некоторое время ничего не происходит, кроме , что процессоры NUMBER_OF_CREATE_KNOWLEDGE_THREADS вращаются на 100% вечно. Что я делаю неправильно? Если вам нужна более конкретная информация, я с удовольствием предоставлю ее для вас!

С уважением, Daniel

Answer 1

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

while(!Thread.currentThread.isInterrupted()) {

}

или

Thread.sleep(0);

Answer 2

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

Answer 3

Вы уверены, что отправленные вами задания действительно завершены?Если вы проверите API для shutdownNow () и shutdown () , вы увидите, что они не гарантируют завершение.

Вы пытались использовать вызов awaitTermination (long timeout, TimeUnit unit) с разумным количеством времени в качестве параметра тайм-аута?(редактировать: «разумное количество времени» зависит, конечно, от среднего времени обработки ваших задач, а также от количества задач, выполняемых в момент, когда вы вызываете завершение)

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

try {

        this.started = true;
        pool.execute(new QueryingAction(pcqs));
        for(;;){
            MyObj p = bq.poll(timeout, TimeUnit.MINUTES); // poll from a blocking queue

            if(p != null){
                if (p.getId().equals("0"))
                    break;

                pool.submit(new AnalysisAction(ds, p, analyzedObjs));
            }else 
                drc.log("Timed out while waiting...");

        }

      } catch (Exception ex) {
          ex.printStackTrace();

      }finally{
          drc.log("--DEBUG: Termination criteria found, shutdown initiated..");
          pool.shutdown();

          int mins = 2;
          int nCores = poolSize -1 ;
          long  totalTasks = pool.getTaskCount(), 
                compTasks = pool.getCompletedTaskCount(),
                tasksRemaining = totalTasks - compTasks,
                timeout = mins * tasksRemaining / nCores;

          drc.log(  "--DEBUG: Shutdown commenced, thread pool will terminate once all objects are processed, " +
                    "or will timeout in : " + timeout + " minutes... \n" + compTasks + " of " +  (totalTasks -1) + 
                    " objects have been analyzed so far, " + "mean process time is: " +
                    drc.getMeanProcTimeAsString() + " milliseconds.");

          pool.awaitTermination(timeout, TimeUnit.MINUTES);
      }

Answer 4

executorService.invokeAll

должен возвращаться только после завершения всех задач.Как и future.get() Вы уверены, что вызов executorService.invokeAll(todo); когда-либо возвращается и не блокирует вечно ожидающие завершения задачи?