Как использовать invokeAll(), чтобы все пулы потоков выполняли свою задачу?

Question

ExecutorService pool=Executors.newFixedThreadPool(7); 
     List<Future<Hotel>> future=new ArrayList<Future<Hotel>>(); 
     List<Callable<Hotel>> callList = new ArrayList<Callable<Hotel>>(); 

     for(int i=0;i<=diff;i++){ 

      String str="2013-"+(liDates.get(i).get(Calendar.MONTH)+1)+"-"+liDates.get(i).get(Calendar.DATE); 

      callList.add(new HotelCheapestFare(str)); 

     }  
    future=pool.invokeAll(callList); 
for(int i=0;i<=future.size();i++){ 

     System.out.println("name is:"+future.get(i).get().getName()); 
    } 

Теперь я хочу объединить в invokeAll все задачи, прежде чем попасть в цикл, но когда я запускаю эту программу для цикла запускается на выполнение до этого invokeAll и выдает это исключение:

java.util.concurrent.ExecutionException: java.lang.NullPointerException at 
java.util.concurrent.FutureTask$Sync.innerGet(Unknown Source) at 
java.util.concurrent.FutureTask.get(Unknown Source) at 
com.mmt.freedom.cheapestfare.TestHotel.main(TestHotel.java:6‌​5) 

Caused by: java.lang.NullPointerException at 
com.mmt.freedom.cheapestfare.HotelCheapestFare.getHotelCheap‌estFare(HotelCheapes‌​tFare.java:166) 
at com.mmt.freedom.cheapestfare.HotelCheapestFare.call(HotelChe‌​apestFare.java:219) 
at com.mmt.freedom.cheapestfare.HotelCheapestFare.call(HotelChe‌​apestFare.java:1) 
at java.util.concurrent.FutureTask$Sync.innerRun(Unknown Source) at java.util.concurrent.FutureTask.run(Unknown Source) 
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) atjava.util.concurrent.ThreadPoolExecutor$Worker.run(Unknow‌​n Source) 
at java.lang.Thread.run 

Answer 1

Способ, которым ExecutorService работ является то, что при вызове invokeAll он ждет все задачи для завершения:

Выполнение поставленных задач, возвращая список фьючерсов затаив статус и результаты, когда все завершена е. Future.isDone() истинно для каждого элемента возвращаемого списка. Обратите внимание, что завершенная задача может иметь , либо нормально, либо путем исключения исключений. Результаты этот метод не определены, если данная коллекция модифицирована, а эта операция выполняется. ¹ (курсив)

Что это означает, что ваши задачи все сделано, но некоторые из них могут быть выброшены исключение. Это исключение является частью Future - вызов get вызывает исключение, которое завершается в ExecutionException.

От вас stacktrack

java.util.concurrent.ExecutionException: java.lang.NullPointerException at 
java.util.concurrent.FutureTask$Sync.innerGet(Unknown Source) at 
java.util.concurrent.FutureTask.get(Unknown Source) at 
           ^^^ <-- from get 

Вы можете видеть, что это действительно так. Одна из ваших задач потерпела неудачу с NPE. ExecutorService поймал исключение и расскажет вам об этом, бросив ExecutionException, когда вы позвоните Future.get.

Теперь, если вы хотите взять задания по мере их заполнения вам нужен ExecutorCompletionService. Это действует как BlockingQueue, что позволит вам опросить задачи по мере их завершения.

public static void main(String[] args) throws Exception { 
    final ExecutorService executorService = Executors.newFixedThreadPool(10); 
    final ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executorService); 
    executorService.submit(new Runnable() { 
     @Override 
     public void run() { 
      for (int i = 0; i < 100; ++i) { 
       try { 
        final Future<String> myValue = completionService.take(); 
        //do stuff with the Future 
        final String result = myValue.get(); 
        System.out.println(result); 
       } catch (InterruptedException ex) { 
        return; 
       } catch (ExecutionException ex) { 
        System.err.println("TASK FAILED"); 
       } 
      } 
     } 
    }); 
    for (int i = 0; i < 100; ++i) { 
     completionService.submit(new Callable<String>() { 
      @Override 
      public String call() throws Exception { 
       if (Math.random() > 0.5) { 
        throw new RuntimeException("FAILED"); 
       } 
       return "SUCCESS"; 
      } 
     }); 
    } 
    executorService.shutdown(); 
} 

В этом примере у меня есть одна задача, которая вызывает take на ExecutorCompletionService который получает Future S, как они становятся доступными, а затем я представляю задачи в ExecutorCompletionService.

Это позволит вам получить неудачную задачу, как только она потерпит неудачу, а не ждать, пока все задачи не сработают вместе.

Единственное осложнение состоит в том, что сложно сказать потоку опроса, что все задачи выполняются, поскольку все теперь асинхронно. В этом случае я использовал знание о том, что будет отправлено 100 заданий, так что ему нужно будет опросить 100 раз. Более общий способ - собрать Future s из метода submit, а затем проложить их, чтобы увидеть, все ли выполнено.

Answer 2

invokeAll является блокирующим методом. Это означает, что JVM не перейдет к следующей строке, пока все потоки не будут завершены. Поэтому я думаю, что что-то не так с вашим будущим результатом потока.

System.out.println("name is:"+future.get(i).get().getName()); 

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

Answer 3

Future.get() броски за исключениями.

CancellationException - если расчет был отменен

ExecutionException - если вычисление сгенерировала исключение

InterruptedException - если текущий поток был прерван во время ожидания

поймать все эти исключения при вызове get() метод ,

Я смоделировали деление на ноль исключения для некоторых Callable задач, но за исключением в одном Callable не влияет на другие Callable задачи, представленные ExecutorService если вы ловите выше трех исключений, как показано в примере кода.

Пример фрагмента кода:

import java.util.concurrent.*; 
import java.util.*; 

public class InvokeAllUsage{ 
    public InvokeAllUsage(){ 
     System.out.println("creating service"); 
     ExecutorService service = Executors.newFixedThreadPool(10); 

     List<MyCallable> futureList = new ArrayList<MyCallable>(); 
     for (int i=0; i<10; i++){ 
      MyCallable myCallable = new MyCallable((long)i+1); 
      futureList.add(myCallable); 
     } 
     System.out.println("Start"); 
     try{ 
      List<Future<Long>> futures = service.invokeAll(futureList); 
      for(Future<Long> future : futures){ 
       try{ 
        System.out.println("future.isDone = " + future.isDone()); 
        System.out.println("future: call ="+future.get()); 
       } 
       catch (CancellationException ce) { 
        ce.printStackTrace(); 
       } catch (ExecutionException ee) { 
        ee.printStackTrace(); 
       } catch (InterruptedException ie) { 
        Thread.currentThread().interrupt(); // ignore/reset 
       } 
      } 
     }catch(Exception err){ 
      err.printStackTrace(); 
     } 
     System.out.println("Completed"); 
     service.shutdown(); 
    } 
    public static void main(String args[]){ 
     InvokeAllUsage demo = new InvokeAllUsage(); 
    } 
    class MyCallable implements Callable<Long>{ 
     Long id = 0L; 
     public MyCallable(Long val){ 
      this.id = val; 
     } 
     public Long call(){ 

      if (id % 5 == 0){ 
       id = id/0; 
      }   
      return id; 
     } 
    } 
} 

выход:

creating service 
Start 
future.isDone = true 
future: call =1 
future.isDone = true 
future: call =2 
future.isDone = true 
future: call =3 
future.isDone = true 
future: call =4 
future.isDone = true 
java.util.concurrent.ExecutionException: java.lang.ArithmeticException:/by zero 
     at java.util.concurrent.FutureTask.report(FutureTask.java:122) 
     at java.util.concurrent.FutureTask.get(FutureTask.java:188) 
     at InvokeAllUsage.<init>(InvokeAllUsage.java:20) 
     at InvokeAllUsage.main(InvokeAllUsage.java:37) 
Caused by: java.lang.ArithmeticException:/by zero 
     at InvokeAllUsage$MyCallable.call(InvokeAllUsage.java:47) 
     at InvokeAllUsage$MyCallable.call(InvokeAllUsage.java:39) 
     at java.util.concurrent.FutureTask.run(FutureTask.java:262) 
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
     at java.lang.Thread.run(Thread.java:744) 
future.isDone = true 
future: call =6 
future.isDone = true 
future: call =7 
future.isDone = true 
future: call =8 
future.isDone = true 
future: call =9 
future.isDone = true 
java.util.concurrent.ExecutionException: java.lang.ArithmeticException:/by zero 
     at java.util.concurrent.FutureTask.report(FutureTask.java:122) 
     at java.util.concurrent.FutureTask.get(FutureTask.java:188) 
     at InvokeAllUsage.<init>(InvokeAllUsage.java:20) 
     at InvokeAllUsage.main(InvokeAllUsage.java:37) 
Caused by: java.lang.ArithmeticException:/by zero 
     at InvokeAllUsage$MyCallable.call(InvokeAllUsage.java:47) 
     at InvokeAllUsage$MyCallable.call(InvokeAllUsage.java:39) 
     at java.util.concurrent.FutureTask.run(FutureTask.java:262) 
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
     at java.lang.Thread.run(Thread.java:744) 
Completed