In Java, wie verarbeite ich CompletableFutures und bekomme das erste erwünschte Ergebnis, das vervollständigt wird?

Question

Normalerweise würde ich mit einer CompleableFuture dann apply oder eine andere Methode aufrufen, etwas zu tun, sobald das Ergebnis verfügbar ist. Allerdings habe ich jetzt eine Situation, in der ich Ergebnisse verarbeiten möchte, bis ich ein positives Ergebnis erhalte und dann alle weiteren Ergebnisse ignoriere.

Wenn ich nur das erste verfügbare Ergebnis nehmen wollte, konnte ich CompletableFuture.anyOf verwenden (obwohl ich es hasse, eine Liste in ein Array konvertieren zu müssen, um anyOf aufzurufen). Aber das will ich nicht. Ich möchte das erste Ergebnis ziehen und wenn es kein wünschenswertes Ergebnis hat, dann möchte ich das zweite verfügbare Ergebnis bearbeiten, bis ich ein wünschenswertes Ergebnis erhalte.

Hier ist ein einfaches Beispiel, das alle Ergebnisse geht durch und gibt den ersten Wert, der es findet, die als 9 größer ist (Beachten Sie, dass dies nicht meine eigentliche Aufgabe ist. Dies ist nur ein einfaches Beispiel.)

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    for(CompletableFuture<Integer> result : results) { 
     Integer v = result.get(); 
     if(v > 9) 
      return v; 
    } 
    return null; 
} 

Natürlich durchläuft dieses Beispiel die Ergebnisse von Anfang an, nicht indem man die Ergebnisse betrachtet, wenn sie vollständig sind. Also hier ist einer, der das erreicht, was ich will, aber mit viel komplizierterem Code.

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    AtomicInteger finalResult = new AtomicInteger(); 
    CountDownLatch latch = new CountDownLatch(results.size()); 
    for(CompletableFuture<Integer> result : results) { 
     result.whenComplete((v,e) -> { 
      if(e!=null) { 
       Logger.getLogger(getClass()).error("",e); 
      } else if(v > 9) { 
       finalResult.set(v); 
       while(latch.getCount() > 0) 
        latch.countDown(); 
       return; 
      } 
      latch.countDown(); 
     }); 
    } 
    latch.await(); 

    if(finalResult.get() > 9) 
     return finalResult.get(); 
    return null; 
}  

Gibt es eine API, wo ich das tun kann?

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    Iterator<Integer> resultIt = getResultsAsAvailable(results); 
    for(; resultIt.hasNext();) { 
     Integer v = resultIt.next(); 
     if(v > 9) 
      return v; 
    } 
    return null; 
} 

Oder noch besser:

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    return getFirstMatch(results, r -> {return r > 9;}); 
} 

Answer 1

Sie können die folgende Lösung verwenden:

public static <T> CompletableFuture<T> anyMatch(
    List<? extends CompletionStage<? extends T>> l, Predicate<? super T> criteria) { 

    CompletableFuture<T> result=new CompletableFuture<>(); 
    Consumer<T> whenMatching=v -> { if(criteria.test(v)) result.complete(v); }; 
    CompletableFuture.allOf(l.stream() 
     .map(f -> f.thenAccept(whenMatching)).toArray(CompletableFuture<?>[]::new)) 
    .whenComplete((ignored, t) -> 
     result.completeExceptionally(t!=null? t: new NoSuchElementException())); 
    return result; 
} 

Das Grundprinzip das gleiche wie in Pillar’s answer ist, gibt es jedoch einige Unterschiede:

• Die generische Signatur flexibler ist.
• Die Erstellung des Arrays erforderlich für CompletableFuture.allOf ist mit der Registrierung der Folgeaktion zu den Source-Futures kombiniert. Als Nebeneffekt ist der Handler der Aktion allOf abhängig von der Beendigung aller Versuche, das Ergebnis abzuschließen, und nicht nur von den ursprünglichen Futures. Dies macht die tatsächlich gewünschte Abhängigkeit explizit. Auf diese Weise würde es sogar funktionieren, wenn wir alle thenAccept s durch thenAcceptAsync s ersetzen.
• Diese Lösung wird mit einem NoSuchElementException statt null in dem Fall abgeschlossen, in dem kein Ergebnis die Kriterien erfüllt. Wenn mindestens eine Zukunft in Ausnahmefällen abgeschlossen wurde und keine erfolgreiche Beendigung mit einem übereinstimmenden Ergebnis vorliegt, wird eine der aufgetretenen Ausnahmen weitergeleitet.

Sie können es versuchen, mit

List<CompletableFuture<Integer>> list=Arrays.asList(
    CompletableFuture.supplyAsync(()->5), 
    CompletableFuture.supplyAsync(()->{throw new RuntimeException(); }), 
    CompletableFuture.supplyAsync(()->42), 
    CompletableFuture.completedFuture(0) 
); 
anyMatch(list, i -> i>9) 
    .thenAccept(i->System.out.println("got "+i)) 
    // optionally chain with: 
    .whenComplete((x,t)->{ if(t!=null) t.printStackTrace(); }); 

Answer 2

Ich weiß nicht, von einer solchen API im JDK oder anderswo. Sie können Ihre eigenen rollen.

Sie können die Tatsache nutzen, dass CompletableFuture#complete (und completeExceptionally) nichts tut, wenn die Zukunft bereits abgeschlossen ist.

Wenn nicht bereits abgeschlossen ist, setzt die von get() und Methoden auf den angegebenen Wert zurückgegebene Wert bezogen.

Erstellen Sie ein neues EndergebnisCompletableFuture. Fügen Sie eine Fortsetzung zu jeder Ihrer Zukünfte hinzu, die versucht, complete dieses Endergebnis wenn Ihre Bedingung zutrifft. Diese Zukunft wird mit dem ersten Erfolg abgeschlossen sein. Wenn jedoch keiner erfolgreich ist, benötigen Sie als Ergebnis null. Sie können ein CompletableFuture mit allOf auch versuchen, complete das Endergebnis mit null zu versuchen.

So etwas wie

public static <T> CompletableFuture<T> firstOrNull(List<CompletableFuture<T>> futures, Predicate<T> condition) { 
    CompletableFuture<T> finalResult = new CompletableFuture<>(); 
    // attempt to complete on success 
    futures.stream().forEach(future -> future.thenAccept(successResult -> { 
     if (condition.test(successResult)) 
      finalResult.complete(successResult); 
    })); 
    CompletableFuture<?> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); 
    all.thenRun(() -> { 
     finalResult.complete(null); 
    }); 
    return finalResult; 
} 

Sie zahlen den Overhead von No-op Anrufungen.

Sie können null auf einen bestimmten Standardwert ändern oder Ausnahmen anders behandeln (completeExceptionally, sobald ein Fehler auftritt). Sie müssen whenComplete oder handle anstelle der thenAccept oben verwenden, um Zugriff auf die Exception zu erhalten.