Warten auf eine Liste der Zukunft

145

Ich habe eine Methode, die eine Listder Futures zurückgibt

List<Future<O>> futures = getFutures();

Jetzt möchte ich warten, bis entweder alle Futures erfolgreich verarbeitet wurden oder eine der Aufgaben, deren Ausgabe von einer Zukunft zurückgegeben wird, eine Ausnahme auslöst. Selbst wenn eine Aufgabe eine Ausnahme auslöst, macht es keinen Sinn, auf die anderen Zukünfte zu warten.

Einfacher Ansatz wäre zu

wait() {

   For(Future f : futures) {
     try {
       f.get();
     } catch(Exception e) {
       //TODO catch specific exception
       // this future threw exception , means somone could not do its task
       return;
     }
   }
}

Aber das Problem hier ist, wenn zum Beispiel die 4. Zukunft eine Ausnahme auslöst, dann werde ich unnötig warten, bis die ersten 3 Futures verfügbar sind.

Wie kann man das lösen? Wird der Countdown in irgendeiner Weise helfen? Ich kann Future nicht verwenden, isDoneda das Java-Dokument dies sagt

boolean isDone()
Returns true if this task completed. Completion may be due to normal termination, an exception, or cancellation -- in all of these cases, this method will return true.
user93796
quelle
1
Wer generiert diese Futures? Von welcher Art sind sie? Die Schnittstelle java.util.concurrent.Future bietet nicht die gewünschte Funktionalität. Die einzige Möglichkeit besteht darin, Ihre eigenen Futures mit Rückrufen zu verwenden.
Alexei Kaigorodov
Sie könnten ExecutionServicefür jeden "Stapel" von Aufgaben eine Instanz erstellen, diese an ihn senden, dann den Dienst sofort herunterfahren und awaitTermination()ihn vermutlich verwenden.
Millimoose
Sie können a verwenden, CountDownLatchwenn Sie den Körper aller Ihrer Futures in a eingewickelt haben try..finally, um sicherzustellen, dass der Riegel ebenfalls dekrementiert wird.
Millimoose
docs.oracle.com/javase/7/docs/api/java/util/concurrent/… macht genau das, was Sie brauchen.
Assylias
@AlexeiKaigorodov JA, meine Zukunft ist vom Typ java.util.concurrent.Ich verklage Zukunft mit callable.Ich bekomme Futture, wenn ich eine Aufgabe an einen Execureservice sende
user93796

Antworten:

124

Sie können einen CompletionService verwenden , um die Futures zu erhalten, sobald sie bereit sind. Wenn einer von ihnen eine Ausnahme auslöst, brechen Sie die Verarbeitung ab. Etwas wie das:

Executor executor = Executors.newFixedThreadPool(4);
CompletionService<SomeResult> completionService = 
       new ExecutorCompletionService<SomeResult>(executor);

//4 tasks
for(int i = 0; i < 4; i++) {
   completionService.submit(new Callable<SomeResult>() {
       public SomeResult call() {
           ...
           return result;
       }
   });
}

int received = 0;
boolean errors = false;

while(received < 4 && !errors) {
      Future<SomeResult> resultFuture = completionService.take(); //blocks if none available
      try {
         SomeResult result = resultFuture.get();
         received ++;
         ... // do something with the result
      }
      catch(Exception e) {
             //log
         errors = true;
      }
}

Ich denke, Sie können sich weiter verbessern, um noch laufende Aufgaben abzubrechen, wenn einer von ihnen einen Fehler auslöst.

dcernahoschi
quelle
1
: Ihr Code hat das gleiche Problem, das ich in meinem Beitrag erwähnt habe. Wenn in Zukunft eine Ausnahme ausgelöst wird, wartet der Code immer noch auf die Fertigstellung von 1,2,3. oder wird vervollständigungSerice.take) die Zukunft zurückgeben, die zuerst abgeschlossen wird?
user93796
1
Was ist mit Zeitüberschreitungen? Kann ich dem Abschlussdienst mitteilen, dass er maximal X Sekunden warten soll?
user93796
1
Sollte nicht haben. Es iteriert nicht über die Futures, aber sobald eine fertig ist, wird sie verarbeitet / verifiziert, wenn keine Ausnahme ausgelöst wird.
dcernahoschi
2
Um eine Zeitüberschreitung zu erzielen, die darauf wartet, dass eine Zukunft in der Warteschlange angezeigt wird, gibt es eine Abfragemethode (Sekunden) für die CompletionService.
dcernahoschi
Hier ist das Arbeitsbeispiel für Github: github.com/princegoyal1987/FutureDemo
user18853
107

Wenn Sie Java 8 verwenden , können Sie dies einfacher mit CompletableFuture und CompletableFuture.allOf tun , die den Rückruf erst anwenden , nachdem alle bereitgestellten CompletableFutures ausgeführt wurden.

// Waits for *all* futures to complete and returns a list of results.
// If *any* future completes exceptionally then the resulting future will also complete exceptionally.

public static <T> CompletableFuture<List<T>> all(List<CompletableFuture<T>> futures) {
    CompletableFuture[] cfs = futures.toArray(new CompletableFuture[futures.size()]);

    return CompletableFuture.allOf(cfs)
            .thenApply(ignored -> futures.stream()
                                    .map(CompletableFuture::join)
                                    .collect(Collectors.toList())
            );
}
Andrejs
quelle
3
Hallo @Andrejs, könnten Sie bitte erklären, was dieser Codeausschnitt bewirkt? Ich sehe dies an mehreren Stellen vorgeschlagen, bin aber verwirrt darüber, was tatsächlich passiert. Wie werden Ausnahmen behandelt, wenn einer der Threads ausfällt?
VSEWHGHP
2
@VSEWHGHP Aus dem Javadoc: Wenn eine der angegebenen CompletableFutures ausnahmsweise abgeschlossen ist, wird dies auch von der zurückgegebenen CompletableFuture ausgeführt, wobei eine CompletionException diese Ausnahme als Ursache enthält.
Andrejs
1
Gibt es eine Möglichkeit, dieses Snippet zu verwenden, aber die Werte für alle anderen Threads zu erhalten, die erfolgreich abgeschlossen wurden? Sollte ich einfach die CompletableFutures-Liste durchlaufen und aufrufen, um die CompletableFuture <List <T >> zu ignorieren, da die Sequenzfunktion dafür sorgt, dass alle Threads entweder mit Ergebnis oder Ausnahme vollständig sind?
VSEWHGHP
6
Dies löst ein anderes Problem. Wenn Sie FutureInstanzen haben, können Sie diese Methode nicht anwenden. Es ist nicht einfach, Futurein zu konvertieren CompletableFuture.
Jarekczek
Es wird nicht funktionieren, wenn wir bei einer Aufgabe eine Ausnahme haben.
Slisnychyi
21

Verwenden Sie a CompletableFuturein Java 8

    // Kick of multiple, asynchronous lookups
    CompletableFuture<User> page1 = gitHubLookupService.findUser("Test1");
    CompletableFuture<User> page2 = gitHubLookupService.findUser("Test2");
    CompletableFuture<User> page3 = gitHubLookupService.findUser("Test3");

    // Wait until they are all done
    CompletableFuture.allOf(page1,page2,page3).join();

    logger.info("--> " + page1.get());
sendon1982
quelle
1
Dies sollte die akzeptierte Antwort sein. Es ist auch Teil der offiziellen Spring-Dokumentation: spring.io/guides/gs/async-method
maaw
Funktioniert wie erwartet.
Dimon
15

Sie können einen ExecutorCompletionService verwenden . Die Dokumentation enthält sogar ein Beispiel für Ihren genauen Anwendungsfall:

Angenommen, Sie möchten stattdessen das erste Nicht-Null-Ergebnis der Aufgabengruppe verwenden, alle Ausnahmen ignorieren und alle anderen Aufgaben abbrechen, wenn die erste fertig ist:

void solve(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException {
    CompletionService<Result> ecs = new ExecutorCompletionService<Result>(e);
    int n = solvers.size();
    List<Future<Result>> futures = new ArrayList<Future<Result>>(n);
    Result result = null;
    try {
        for (Callable<Result> s : solvers)
            futures.add(ecs.submit(s));
        for (int i = 0; i < n; ++i) {
            try {
                Result r = ecs.take().get();
                if (r != null) {
                    result = r;
                    break;
                }
            } catch (ExecutionException ignore) {
            }
        }
    } finally {
        for (Future<Result> f : futures)
            f.cancel(true);
    }

    if (result != null)
        use(result);
}

Wichtig hierbei ist, dass ecs.take () die erste abgeschlossene Aufgabe erhält , nicht nur die erste eingereichte. Daher sollten Sie sie in der Reihenfolge erhalten, in der die Ausführung beendet ist (oder eine Ausnahme ausgelöst wird).

jmiserez
quelle
3

Wenn Sie Java 8 verwenden und CompletableFutures nicht manipulieren möchten , habe ich ein Tool geschrieben, mit dem Sie Ergebnisse für ein List<Future<T>>Streaming abrufen können . Der Schlüssel ist, dass es dir verboten ist, map(Future::get)wenn es wirft.

public final class Futures
{

    private Futures()
    {}

    public static <E> Collector<Future<E>, Collection<E>, List<E>> present()
    {
        return new FutureCollector<>();
    }

    private static class FutureCollector<T> implements Collector<Future<T>, Collection<T>, List<T>>
    {
        private final List<Throwable> exceptions = new LinkedList<>();

        @Override
        public Supplier<Collection<T>> supplier()
        {
            return LinkedList::new;
        }

        @Override
        public BiConsumer<Collection<T>, Future<T>> accumulator()
        {
            return (r, f) -> {
                try
                {
                    r.add(f.get());
                }
                catch (InterruptedException e)
                {}
                catch (ExecutionException e)
                {
                    exceptions.add(e.getCause());
                }
            };
        }

        @Override
        public BinaryOperator<Collection<T>> combiner()
        {
            return (l1, l2) -> {
                l1.addAll(l2);
                return l1;
            };
        }

        @Override
        public Function<Collection<T>, List<T>> finisher()
        {
            return l -> {

                List<T> ret = new ArrayList<>(l);
                if (!exceptions.isEmpty())
                    throw new AggregateException(exceptions, ret);

                return ret;
            };

        }

        @Override
        public Set<java.util.stream.Collector.Characteristics> characteristics()
        {
            return java.util.Collections.emptySet();
        }
    }

Dies erfordert eine AggregateException, die wie C # funktioniert

public class AggregateException extends RuntimeException
{
    /**
     *
     */
    private static final long serialVersionUID = -4477649337710077094L;

    private final List<Throwable> causes;
    private List<?> successfulElements;

    public AggregateException(List<Throwable> causes, List<?> l)
    {
        this.causes = causes;
        successfulElements = l;
    }

    public AggregateException(List<Throwable> causes)
    {
        this.causes = causes;
    }

    @Override
    public synchronized Throwable getCause()
    {
        return this;
    }

    public List<Throwable> getCauses()
    {
        return causes;
    }

    public List<?> getSuccessfulElements()
    {
        return successfulElements;
    }

    public void setSuccessfulElements(List<?> successfulElements)
    {
        this.successfulElements = successfulElements;
    }

}

Diese Komponente fungiert genau wie die Task.WaitAll von C # . Ich arbeite an einer Variante, die das Gleiche tut wie CompletableFuture.allOf(äquivalent zu Task.WhenAll)

Der Grund, warum ich das getan habe, ist, dass ich Spring's verwende ListenableFutureund nicht portieren möchte, CompletableFutureobwohl dies ein Standardweg ist

usr-local-ΕΨΗΕΛΩΝ
quelle
1
Upvote für die Notwendigkeit einer gleichwertigen AggregateException.
GranadaCoder
Ein Beispiel für die Verwendung dieser Einrichtung wäre schön.
XDS
1

Wenn Sie eine Liste von CompletableFutures kombinieren möchten, können Sie dies tun:

List<CompletableFuture<Void>> futures = new ArrayList<>();
// ... Add futures to this ArrayList of CompletableFutures

// CompletableFuture.allOf() method demand a variadic arguments
// You can use this syntax to pass a List instead
CompletableFuture<Void> allFutures = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[futures.size()]));

// Wait for all individual CompletableFuture to complete
// All individual CompletableFutures are executed in parallel
allFutures.get();

Weitere Informationen zu Future & CompletableFuture finden Sie unter nützlichen Links:
1. Future: https://www.baeldung.com/java-future
2. CompletableFuture: https://www.baeldung.com/java-completablefuture
3. CompletableFuture: https : //www.callicoder.com/java-8-completablefuture-tutorial/

Bohao LI
quelle
0

Vielleicht würde dies helfen (nichts würde durch rohen Thread ersetzt, ja!). Ich schlage vor, jeden FutureKerl mit einem getrennten Thread auszuführen (sie gehen parallel). Wenn dann einer der Fehler auftritt, signalisiert dies nur dem Manager (der HandlerKlasse).

class Handler{
//...
private Thread thisThread;
private boolean failed=false;
private Thread[] trds;
public void waitFor(){
  thisThread=Thread.currentThread();
  List<Future<Object>> futures = getFutures();
  trds=new Thread[futures.size()];
  for (int i = 0; i < trds.length; i++) {
    RunTask rt=new RunTask(futures.get(i), this);
    trds[i]=new Thread(rt);
  }
  synchronized (this) {
    for(Thread tx:trds){
      tx.start();
    }  
  }
  for(Thread tx:trds){
    try {tx.join();
    } catch (InterruptedException e) {
      System.out.println("Job failed!");break;
    }
  }if(!failed){System.out.println("Job Done");}
}

private List<Future<Object>> getFutures() {
  return null;
}

public synchronized void cancelOther(){if(failed){return;}
  failed=true;
  for(Thread tx:trds){
    tx.stop();//Deprecated but works here like a boss
  }thisThread.interrupt();
}
//...
}
class RunTask implements Runnable{
private Future f;private Handler h;
public RunTask(Future f,Handler h){this.f=f;this.h=h;}
public void run(){
try{
f.get();//beware about state of working, the stop() method throws ThreadDeath Error at any thread state (unless it blocked by some operation)
}catch(Exception e){System.out.println("Error, stopping other guys...");h.cancelOther();}
catch(Throwable t){System.out.println("Oops, some other guy has stopped working...");}
}
}

Ich muss sagen, dass der obige Code fehlerhaft wäre (nicht überprüft), aber ich hoffe, ich könnte die Lösung erklären. Bitte probieren Sie es aus.


quelle
0
 /**
     * execute suppliers as future tasks then wait / join for getting results
     * @param functors a supplier(s) to execute
     * @return a list of results
     */
    private List getResultsInFuture(Supplier<?>... functors) {
        CompletableFuture[] futures = stream(functors)
                .map(CompletableFuture::supplyAsync)
                .collect(Collectors.toList())
                .toArray(new CompletableFuture[functors.length]);
        CompletableFuture.allOf(futures).join();
        return stream(futures).map(a-> {
            try {
                return a.get();
            } catch (InterruptedException | ExecutionException e) {
                //logger.error("an error occurred during runtime execution a function",e);
                return null;
            }
        }).collect(Collectors.toList());
    };
Mohamed.Abdo
quelle
0

Der CompletionService nimmt Ihre Callables mit der .submit () -Methode und Sie können die berechneten Futures mit der .take () -Methode abrufen.

Eine Sache, die Sie nicht vergessen dürfen, ist das Beenden des ExecutorService durch Aufrufen der .shutdown () -Methode. Sie können diese Methode auch nur aufrufen, wenn Sie einen Verweis auf den Executor-Service gespeichert haben. Stellen Sie daher sicher, dass Sie einen behalten.

Beispielcode - Für eine feste Anzahl von Arbeitselementen, die parallel bearbeitet werden sollen:

ExecutorService service = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

CompletionService<YourCallableImplementor> completionService = 
new ExecutorCompletionService<YourCallableImplementor>(service);

ArrayList<Future<YourCallableImplementor>> futures = new ArrayList<Future<YourCallableImplementor>>();

for (String computeMe : elementsToCompute) {
    futures.add(completionService.submit(new YourCallableImplementor(computeMe)));
}
//now retrieve the futures after computation (auto wait for it)
int received = 0;

while(received < elementsToCompute.size()) {
 Future<YourCallableImplementor> resultFuture = completionService.take(); 
 YourCallableImplementor result = resultFuture.get();
 received ++;
}
//important: shutdown your ExecutorService
service.shutdown();

Beispielcode - Für eine dynamische Anzahl von Arbeitselementen, die parallel bearbeitet werden sollen:

public void runIt(){
    ExecutorService service = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    CompletionService<CallableImplementor> completionService = new ExecutorCompletionService<CallableImplementor>(service);
    ArrayList<Future<CallableImplementor>> futures = new ArrayList<Future<CallableImplementor>>();

    //Initial workload is 8 threads
    for (int i = 0; i < 9; i++) {
        futures.add(completionService.submit(write.new CallableImplementor()));             
    }
    boolean finished = false;
    while (!finished) {
        try {
            Future<CallableImplementor> resultFuture;
            resultFuture = completionService.take();
            CallableImplementor result = resultFuture.get();
            finished = doSomethingWith(result.getResult());
            result.setResult(null);
            result = null;
            resultFuture = null;
            //After work package has been finished create new work package and add it to futures
            futures.add(completionService.submit(write.new CallableImplementor()));
        } catch (InterruptedException | ExecutionException e) {
            //handle interrupted and assert correct thread / work packet count              
        } 
    }

    //important: shutdown your ExecutorService
    service.shutdown();
}

public class CallableImplementor implements Callable{
    boolean result;

    @Override
    public CallableImplementor call() throws Exception {
        //business logic goes here
        return this;
    }

    public boolean getResult() {
        return result;
    }

    public void setResult(boolean result) {
        this.result = result;
    }
}
fl0w
quelle
0

Ich habe eine Utility-Klasse, die Folgendes enthält:

@FunctionalInterface
public interface CheckedSupplier<X> {
  X get() throws Throwable;
}

public static <X> Supplier<X> uncheckedSupplier(final CheckedSupplier<X> supplier) {
    return () -> {
        try {
            return supplier.get();
        } catch (final Throwable checkedException) {
            throw new IllegalStateException(checkedException);
        }
    };
}

Sobald Sie das haben, können Sie mit einem statischen Import einfach auf alle Futures wie diese warten:

futures.stream().forEach(future -> uncheckedSupplier(future::get).get());

Sie können auch alle Ergebnisse wie folgt sammeln:

List<MyResultType> results = futures.stream()
    .map(future -> uncheckedSupplier(future::get).get())
    .collect(Collectors.toList());

Ich besuche nur meinen alten Beitrag und bemerke, dass du noch einen Kummer hattest:

Aber das Problem hier ist, wenn zum Beispiel die 4. Zukunft eine Ausnahme auslöst, dann werde ich unnötig warten, bis die ersten 3 Futures verfügbar sind.

In diesem Fall besteht die einfache Lösung darin, dies parallel zu tun:

futures.stream().parallel()
 .forEach(future -> uncheckedSupplier(future::get).get());

Auf diese Weise wird die erste Ausnahme, obwohl sie die Zukunft nicht aufhalten wird, die forEach-Anweisung wie im seriellen Beispiel unterbrechen, aber da alle parallel warten, müssen Sie nicht warten, bis die ersten 3 abgeschlossen sind.

Brixomatisch
quelle
0
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class Stack2 {   
    public static void waitFor(List<Future<?>> futures) {
        List<Future<?>> futureCopies = new ArrayList<Future<?>>(futures);//contains features for which status has not been completed
        while (!futureCopies.isEmpty()) {//worst case :all task worked without exception, then this method should wait for all tasks
            Iterator<Future<?>> futureCopiesIterator = futureCopies.iterator();
            while (futureCopiesIterator.hasNext()) {
                Future<?> future = futureCopiesIterator.next();
                if (future.isDone()) {//already done
                    futureCopiesIterator.remove();
                    try {
                        future.get();// no longer waiting
                    } catch (InterruptedException e) {
                        //ignore
                        //only happen when current Thread interrupted
                    } catch (ExecutionException e) {
                        Throwable throwable = e.getCause();// real cause of exception
                        futureCopies.forEach(f -> f.cancel(true));//cancel other tasks that not completed
                        return;
                    }
                }
            }
        }
    }
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        Runnable runnable1 = new Runnable (){
            public void run(){
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                }
            }
        };
        Runnable runnable2 = new Runnable (){
            public void run(){
                try {
                    Thread.sleep(4000);
                } catch (InterruptedException e) {
                }
            }
        };


        Runnable fail = new Runnable (){
            public void run(){
                try {
                    Thread.sleep(1000);
                    throw new RuntimeException("bla bla bla");
                } catch (InterruptedException e) {
                }
            }
        };

        List<Future<?>> futures = Stream.of(runnable1,fail,runnable2)
                .map(executorService::submit)
                .collect(Collectors.toList());

        double start = System.nanoTime();
        waitFor(futures);
        double end = (System.nanoTime()-start)/1e9;
        System.out.println(end +" seconds");

    }
}
Farhad Baghirov
quelle