Come utilizzare invokeAll() per consentire a tutti i pool di thread di eseguire le proprie attività?

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

Ora voglio piscina a invokeAll tutto il compito prima di arrivare al ciclo for, ma quando ho eseguito questo programma per il ciclo viene eseguito prima che invokeAll e genera questa eccezione:

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

Il modo in cui un ExecutorService opere è che quando si chiama invokeAll attende per tutti i compiti da completare:

esegue i compiti assegnati, restituendo una lista di Futures tenendo il loro status di e risultati quando tutti complet e. Future.isDone() è true per ogni elemento dell'elenco restituito. Si noti che un'operazione completata potrebbe avere terminata normalmente o lanciando un'eccezione. I risultati di questo metodo non sono definiti se la raccolta specificata viene modificata mentre questa operazione è in corso. ¹ (enfasi aggiunta)

Ciò significa che le attività sono tutti fatti, ma alcuni possono avere generato un'eccezione. Questa eccezione fa parte di Future - la chiamata a get causa la riconduzione dell'eccezione in un ExecutionException.

Da voi 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 

Si può vedere che questo è davvero il caso. Uno dei tuoi compiti è fallito con un NPE. Lo ExecutorService ha rilevato un'eccezione e lo sta dicendo lanciando un ExecutionException quando si chiama Future.get.

Ora, se si desidera eseguire attività mentre completano è necessario un ExecutorCompletionService. Funziona come un BlockingQueue che ti consentirà di eseguire il polling per le attività man mano che finiscono.

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 questo esempio ho un compito che richiede take sul ExecutorCompletionService che ottiene i Future s non appena saranno disponibili e poi presentare compiti al ExecutorCompletionService.

Ciò consentirà di ottenere l'attività non riuscita non appena fallisce piuttosto che dover attendere che tutte le attività abbiano esito negativo.

L'unica complicazione è che è difficile dire al thread di polling che tutte le attività sono eseguite poiché tutto ora è asincrono. In questo caso ho utilizzato la consapevolezza che verranno inviate 100 attività in modo che sia necessario eseguire il polling 100 volte. Un modo più generale sarebbe quello di raccogliere gli Future s dal metodo submit e quindi ricollegarli per vedere se tutto è stato completato.

Answer 2

invokeAll è un metodo di blocco. Significa che JVM non procederà alla riga successiva finché tutti i thread non saranno completati. Quindi penso che ci sia qualcosa di sbagliato nei risultati futuri del tuo thread.

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

da questa linea penso che ci sono alcuni dei futures non hanno risultato e può essere nullo, così si dovrebbe controllare il codice, se ci sono alcuni Futures nulli, in tal caso, ottenere un se prima di questa linea eseguito.

Answer 3

Future.get() getta sotto le eccezioni.

CancellationException - se il calcolo è stato annullato

ExecutionException - se il calcolo ha generato un'eccezione

InterruptedException - se il thread corrente è stata interrotta in attesa

catturare tutte queste eccezioni quando si chiama get() metodo .

Ho simulato dividere per eccezione zero per alcuni compiti, ma Callable eccezione in una Callable non influisce altri Callable attività sottoposte alla ExecutorService se si cattura sopra tre eccezioni, come mostrato nel codice di esempio. Codice

Esempio frammento:

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

uscita:

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