Wie benutze invokeAll(), damit alle Thread-Pools ihre Aufgabe erledigen?

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()); 
    } 

Jetzt möchte ich bündeln, um invokeAll alle die Aufgabe, bevor sie für Schleife in die Einstiegs-, aber wenn ich dieses Programm für Schleife laufen ausgeführt wird, bevor dieser invokeAll und wirft diese Ausnahme:

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

Die Art und Weise ein ExecutorService Werke das heißt, wenn Sie invokeAll nennen es für alle Aufgaben erledigen wartet:

Führt die gestellten Aufgaben, Zurückgeben einer Liste von Futures ihre Status und die Ergebnisse zu halten, wenn alle complet e. Future.isDone() gilt für jedes Element der zurückgegebenen Liste. Beachten Sie, dass eine abgeschlossene Task entweder normal beendet oder eine Ausnahme auslösen konnte. Die Ergebnisse von diese Methode sind nicht definiert, wenn die angegebene Sammlung geändert wird, während dieser Vorgang ausgeführt wird. ¹ (Hervorhebung hinzugefügt)

Was das bedeutet, ist, dass Ihre Aufgaben alles getan werden, aber einige eine Ausnahme geworfen haben. Diese Ausnahme ist Teil der Future - Aufruf get bewirkt, dass die Ausnahme in einem ExecutionException verpackt umbrochen wird.

Von Ihnen

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 

stacktrack können Sie sehen, dass dies tatsächlich der Fall ist. Eine Ihrer Aufgaben ist mit einer NPE fehlgeschlagen. Die ExecutorService erregte die Ausnahme und erzählt Ihnen davon, indem Sie ExecutionException werfen, wenn Sie Future.get anrufen.

Nun, wenn Sie Aufgaben übernehmen möchten, wie sie abschließen, benötigen Sie eine ExecutorCompletionService. Dies dient als BlockingQueue, mit dem Sie nach und nach Aufgaben abrufen können.

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(); 
} 

In diesem Beispiel habe ich eine Aufgabe, die take auf dem ExecutorCompletionService nennt, die die Future s bekommt, wie sie die ExecutorCompletionService verfügbar sind und ich einreichen Aufgaben dann.

Dies ermöglicht es Ihnen, die fehlgeschlagene Aufgabe zu erhalten, sobald sie fehlschlägt, anstatt zu warten, bis alle Aufgaben gemeinsam fehlschlagen.

Die einzige Komplikation ist, dass es schwierig ist, dem Abfrage-Thread mitzuteilen, dass alle Aufgaben erledigt sind, da alles jetzt asynchron ist. In diesem Fall habe ich das Wissen verwendet, dass 100 Aufgaben eingereicht werden, so dass es nur 100 Mal abfragen muss. Ein allgemeinerer Weg wäre, die Future s von der submit Methode auch zu sammeln und dann über sie zu durchlaufen, um zu sehen, ob alles fertig ist.

Answer 2

invokeAll ist eine Blockierungsmethode. Es bedeutet, dass JVM nicht mit der nächsten Zeile fortfährt, bis alle Threads abgeschlossen sind. Ich denke also, dass etwas mit dem zukünftigen Ergebnis des Threads nicht stimmt.

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

von dieser Linie ist Ich denke, es gibt einige Futures haben kein Ergebnis und kann null sein, so sollten Sie Ihren Code überprüfen, ob es einige Futures null ist, wenn ja, ein, wenn vor dieser Zeile ausgeführt.

Answer 3

Future.get() löst unter Ausnahmen aus.

CancellationException - wenn die Berechnung

ExecutionException abgebrochen wurde - wenn die Berechnung eine Ausnahme ausgelöst hat

InterruptedException - wenn der aktuelle Thread unterbrochen wurde, während

fangen alle diese Ausnahmen warten, wenn Sie get() Methode aufrufen .

Ich habe divide simuliert durch Null Ausnahme für einige Callable Aufgaben aber Ausnahme in einer Callable nicht andere Callable Aufgaben ExecutorService eingereicht nicht beeinträchtigt, wenn Sie über drei Ausnahmen fangen, wie in Beispielcode gezeigt.

Beispiel Code-Schnipsel:

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; 
     } 
    } 
} 

Ausgang:

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