Come attendere la lista di `Future`s creati usando diversi` ExecutorServices`

Question

Ok, quindi so che la prima risposta/commento qui sarà "usa uno ExecutorService e usa invokeAll". Tuttavia, c'è una buona ragione (che non annoierò alle persone) per noi mantenendo separati i pool di thread.

quindi ho un elenco di pool di thread (ExecutorServices) e quello che devo fare è invocare un diverso Callable su ogni pool di thread utilizzando submit (nessun problema). Ora ho questa raccolta di istanze Future, ognuna creata su un ExecutorService separato, e desidero attendere che tutte siano completate (ed essere in grado di fornire un timeout in cui le eventuali non eseguite vengono annullate).

Esiste una classe esistente che eseguirà questa operazione (racchiudere un elenco di istanze di Future e attendere fino a quando non saranno state completate tutte)? In caso contrario, i suggerimenti su un meccanismo efficiente sarebbero apprezzati.

Stavo pensando di chiamare get con un timeout per ciascuno, ma devo fare un calcolo del tempo totale passato per ogni chiamata.

Ho visto questo post Wait Until Any of Future is Done ma questo estende Future invece di avvolgerne un elenco.

Answer 1

commento Per Louis', quello che cercava era Futures.successfulAsList

Questo mi permette di aspettare che tutti completino e poi di controllare eventuali futuri che non hanno funzionato.

GUava REGOLE!

Answer 2

Non credo che JDK fornisca un'API diretta che ti consenta di farlo. Tuttavia, penso che sia ugualmente semplice creare un metodo semplice che faccia questo. Potresti dare un'occhiata all'implementazione di AbstractExecutorService.invokeAll() per avere un'idea che questo può essere fatto.

In sostanza, si chiamerebbe future.get() in ogni futuro, diminuendo il tempo di attesa per il tempo necessario ad attendere il risultato ogni volta, e prima di tornare dal metodo annullare tutti i future in sospeso.

Answer 3

Forse non l'ho capito davvero. Tuttavia, per me sembra ancora semplice come

public <V> List<V> get(List<Future<V>> futures, long timeout, TimeUnit unit) 
      throws InterruptedException, ExecutionException, TimeoutException { 
    List<V> result = new ArrayList<V>(); 
    long end = System.nanoTime() + unit.toNanos(timeout); 
    for (Future<V> f: futures) { 
     result.add(f.get(end - System.nanoTime(), TimeUnit.NANOSECONDS)); 
    } 
    return result; 
} 

Mi sbaglio con quello?

La domanda che si collega è molto più complessa, penso, in quanto vogliono solo aspettare il più veloce e, naturalmente, non hanno idea di quale sarà il più veloce.

Answer 4

Questo potrebbe richiedere un po 'di pulizia, ma dovrebbe risolvere il tuo problema. (Alcuni incapsulamento omesso per tempo e nello spazio):

public static <T> LatchWithWrappedCallables<T> wrapCallables(Collection<Callable<T>> callablesToWrap) 
{ 
    CountDownLatch latch = new CountDownLatch(callablesToWrap.size()); 
    List<Callable<T>> wrapped = new ArrayList<Callable<T>>(callablesToWrap.size()); 
    for (Callable<T> currCallable : callablesToWrap) 
    { 
     wrapped.add(new CallableCountdownWrapper<T>(currCallable, latch)); 
    } 

    LatchWithWrappedCallables<T> returnVal = new LatchWithWrappedCallables<T>(); 
    returnVal.latch = latch; 
    returnVal.wrappedCallables = wrapped; 
    return returnVal; 
} 

public static class LatchWithWrappedCallables<T> 
{ 
    public CountDownLatch latch; 
    public Collection<Callable<T>> wrappedCallables; 
} 

public static class CallableCountdownWrapper<T> implements Callable<T> 
{ 
    private final Callable<T> wrapped; 

    private final CountDownLatch latch; 

    public CallableCountdownWrapper(Callable<T> wrapped, CountDownLatch latch) 
    { 
     this.wrapped = wrapped; 
     this.latch = latch; 
    } 

    @Override 
    public T call() throws Exception 
    { 
     try 
     { 
      return wrapped.call(); 
     } 
     finally 
     { 
      latch.countDown(); 
     } 
    } 
} 

Allora il vostro codice sarebbe chiamare in questo modo:

Collection<Callable<String>> callablesToWrap = [Your callables that you need to wait for here]; 
LatchWithWrappedCallables<String> latchAndCallables = wrapCallables(callablesToWrap); 

[Submit the wrapped callables to the executors here] 

if(latchAndCallables.latch.await(timeToWaitInSec, TimeUnit.SECONDS)) 
{ 
    [Handling for timeout here] 
}