streamalism.org

1. Inleiding

Deze tutorial is een gids voor de functionaliteit en gebruiksscenario's van het CompletableFuture klasse die werd geïntroduceerd als een Java 8 Concurrency API-verbetering.

2. Asynchrone berekening in Java

Asynchrone berekeningen zijn moeilijk te beredeneren. Meestal willen we elke berekening beschouwen als een reeks stappen, maar in het geval van asynchrone berekening, acties die worden weergegeven als callbacks zijn meestal ofwel verspreid over de code of diep in elkaar genest. Het wordt nog erger als we fouten moeten afhandelen die kunnen optreden tijdens een van de stappen.

De Toekomst interface is toegevoegd in Java 5 om te dienen als resultaat van een asynchrone berekening, maar het had geen methoden om deze berekeningen te combineren of mogelijke fouten af ​​te handelen.

Java 8 introduceerde het CompletableFuture klasse. Samen met de Toekomst interface, implementeerde het ook de Voltooiingsfase koppel. Deze interface definieert het contract voor een asynchrone rekenstap die we kunnen combineren met andere stappen.

CompletableFuture is tegelijkertijd een bouwsteen en een raamwerk, met ongeveer 50 verschillende methoden voor het samenstellen, combineren en uitvoeren van asynchrone rekenstappen en het afhandelen van fouten.

Zo'n grote API kan overweldigend zijn, maar deze vallen meestal in verschillende duidelijke en verschillende gebruikssituaties.

3. Met behulp van CompletableFuture als een simpele Toekomst

Allereerst de CompletableFuture class implementeert het Toekomst interface, dus we kunnen gebruik het als een Toekomst implementatie, maar met aanvullende voltooiingslogica.

We kunnen bijvoorbeeld een instantie van deze klasse maken met een constructor zonder argumenten om een ​​toekomstig resultaat weer te geven, deze uitdelen aan de consumenten en deze op een bepaald moment in de toekomst voltooien met de compleet methode. De consumenten kunnen de krijgen methode om de huidige thread te blokkeren totdat dit resultaat wordt geleverd.

In het onderstaande voorbeeld hebben we een methode die een CompletableFuture instantie, draait dan een berekening af in een andere thread en retourneert de Toekomst direct.

Wanneer de berekening is voltooid, voltooit de methode de Toekomst door het resultaat aan de compleet methode:

public Future berekenAsync () gooit InterruptedException {CompletableFuture completableFuture = nieuw CompletableFuture (); Executors.newCachedThreadPool (). Submit (() -> {Thread.sleep (500); completableFuture.complete ("Hallo"); return null;}); return completableFuture; }

Om de berekening uit te voeren, gebruiken we de Uitvoerder API. Deze methode voor het maken en voltooien van een CompletableFuture kan samen met elk gelijktijdigheidsmechanisme of API worden gebruikt, inclusief onbewerkte threads.

Let erop dat de berekenAsync methode retourneert een Toekomst voorbeeld.

We noemen gewoon de methode, ontvangen de Toekomst instantie, en roep het krijgen methode erop als we klaar zijn om te blokkeren voor het resultaat.

Merk ook op dat de krijgen methode gooit enkele gecontroleerde uitzonderingen, namelijk ExecutionException (waarin een uitzondering is opgenomen die tijdens een berekening is opgetreden) en InterruptedException (een uitzondering die aangeeft dat een thread die een methode uitvoert, is onderbroken):

Future completableFuture = berekenAsync (); // ... String resultaat = completableFuture.get (); assertEquals ("Hallo", resultaat);

Als we het resultaat van een berekening al kennenkunnen we de static gebruiken voltooidFuture methode met een argument dat een resultaat van deze berekening vertegenwoordigt. Bijgevolg is het krijgen methode van de Toekomst zal nooit blokkeren, en in plaats daarvan onmiddellijk dit resultaat retourneren:

Future completableFuture = CompletableFuture.completedFuture ("Hallo"); // ... String resultaat = completableFuture.get (); assertEquals ("Hallo", resultaat);

Als alternatief scenario willen we misschien de uitvoering van een Toekomst.

4. CompletableFuture met ingekapselde rekenlogica

De bovenstaande code stelt ons in staat om elk mechanisme van gelijktijdige uitvoering te kiezen, maar wat als we deze standaardplaat willen overslaan en gewoon wat code asynchroon willen uitvoeren?

Statische methoden runAsync en supplyAsync stellen ons in staat om een CompletableFuture instantie uit Runnable en Leverancier functionele types navenant.

Beide Runnable en Leverancier zijn functionele interfaces die het mogelijk maken om hun instanties door te geven als lambda-expressies dankzij de nieuwe Java 8-functie.

De Runnable interface is dezelfde oude interface die wordt gebruikt in threads en het staat niet toe om een ​​waarde te retourneren.

De Leverancier interface is een generieke functionele interface met een enkele methode die geen argumenten heeft en een waarde van een geparametriseerd type retourneert.

Dit stelt ons in staat geef een instantie van de Leverancier als een lambda-uitdrukking die de berekening uitvoert en het resultaat retourneert. Het is zo simpel als:

CompletableFuture future = CompletableFuture.supplyAsync (() -> "Hallo"); // ... assertEquals ("Hallo", future.get ());

5. Verwerking van resultaten van asynchrone berekeningen

De meest algemene manier om het resultaat van een berekening te verwerken, is door het naar een functie te sturen. De dan toepassen methode doet precies dat; het accepteert een Functie instantie, gebruikt het om het resultaat te verwerken en retourneert een Toekomst die een waarde bevat die wordt geretourneerd door een functie:

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo"); CompletableFuture future = completableFuture .thenApply (s -> s + "World"); assertEquals ("Hallo wereld", future.get ());

Als we geen waarde hoeven terug te geven onder de Toekomst chain, kunnen we een instantie van de Klant functionele interface. De enkele methode heeft een parameter en keert terug leegte.

Er is een methode voor dit gebruik in het CompletableFuture. De Accepteer dan methode ontvangt een Klant en geeft het het resultaat van de berekening door. Dan de finale future.get () call retourneert een instantie van de Ongeldig type:

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo"); CompletableFuture future = completableFuture .thenAccept (s -> System.out.println ("Berekening geretourneerd:" + s)); future.get ();

Ten slotte, als we de waarde van de berekening niet nodig hebben, noch een waarde aan het einde van de keten willen retourneren, kunnen we een Runnable lambda naar de dan rennen methode. In het volgende voorbeeld drukken we gewoon een regel af in de console nadat we de future.get ():

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo"); CompletableFuture future = completableFuture .thenRun (() -> System.out.println ("Berekening voltooid.")); future.get ();

6. Combineren van futures

Het beste deel van de CompletableFuture API is de vermogen om te combineren CompletableFuture instanties in een reeks berekeningsstappen.

Het resultaat van deze ketting is zelf een CompletableFuture dat maakt het mogelijk om verder te koppelen en te combineren. Deze benadering is alomtegenwoordig in functionele talen en wordt vaak een monadisch ontwerppatroon genoemd.

In het volgende voorbeeld gebruiken we de dan Componeren methode om twee te ketenen Futures opeenvolgend.

Merk op dat deze methode een functie heeft die een retourneert CompletableFuture voorbeeld. Het argument van deze functie is het resultaat van de vorige berekeningsstap. Hierdoor kunnen we deze waarde binnen de volgende gebruiken CompletableFuture‘S lambda:

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo") .thenCompose (s -> CompletableFuture.supplyAsync (() -> s + "Wereld")); assertEquals ("Hallo wereld", completableFuture.get ());

De dan Componeren methode, samen met dan toepassen, de basisbouwstenen van het monadische patroon implementeren. Ze houden nauw verband met de kaart en flatMap methodes van Stroom en Optioneel klassen ook beschikbaar in Java 8.

Beide methoden krijgen een functie en passen deze toe op het berekeningsresultaat, maar de dan Componeren (flatMap) methode ontvangt een functie die een ander object van hetzelfde type retourneert. Deze functionele structuur maakt het mogelijk om de instanties van deze klassen als bouwstenen samen te stellen.

Als we twee onafhankelijk willen uitvoeren Futures en iets doen met hun resultaten, kunnen we de danCombineer methode die een Toekomst en een Functie met twee argumenten om beide resultaten te verwerken:

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo") .thenCombine (CompletableFuture.supplyAsync (() -> "Wereld"), (s1, s2) -> s1 + s2)); assertEquals ("Hallo wereld", completableFuture.get ());

Een eenvoudiger geval is wanneer we iets met twee willen doen Futures‘Resultaten, maar u hoeft geen resulterende waarde door te geven naar een Toekomst ketting. De danAcceptBoth methode is er om te helpen:

CompletableFuture future = CompletableFuture.supplyAsync (() -> "Hallo") .thenAcceptBoth (CompletableFuture.supplyAsync (() -> "Wereld"), (s1, s2) -> System.out.println (s1 + s2));

7. Verschil tussen dan toepassen () en thenCompose ()

In onze vorige secties hebben we voorbeelden laten zien met betrekking tot dan toepassen () en thenCompose (). Beide API's helpen anders te ketenen CompletableFuture oproepen, maar het gebruik van deze 2 functies is anders.

7.1. dan toepassen ()

We kunnen deze methode gebruiken om te werken met een resultaat van de vorige aanroep. Een belangrijk punt om te onthouden is echter dat het retourtype wordt gecombineerd met alle oproepen.

Deze methode is dus handig als we het resultaat van a willen transformeren CompletableFuture bellen:

CompletableFuture finalResult = compute (). ThenApply (s-> s + 1);

7.2. thenCompose ()

De thenCompose () methode is vergelijkbaar met dan toepassen () in die zin dat beide een nieuwe voltooiingsfase retourneren. Echter, thenCompose () gebruikt de vorige fase als argument. Het zal afvlakken en een Toekomst met het resultaat direct, in plaats van een geneste toekomst zoals we hebben waargenomen in thenApply ():

CompletableFuture computeAnother (Integer i) {return CompletableFuture.supplyAsync (() -> 10 + i); } CompletableFuture finalResult = compute (). ThenCompose (this :: computeAnother);

Dus als het idee is om te ketenen CompletableFuture methoden dan is het beter om te gebruiken thenCompose ().

Merk ook op dat het verschil tussen deze twee methoden analoog is aan het verschil tussen kaart() en flatMap ().

8. Meerdere draaien Futures parallel

Wanneer we meerdere moeten uitvoeren Futures parallel willen we meestal wachten tot ze allemaal zijn uitgevoerd en vervolgens hun gecombineerde resultaten verwerken.

De CompletableFuture.allOf statische methode laat toe om te wachten op voltooiing van alle Futures geleverd als een var-arg:

CompletableFuture future1 = CompletableFuture.supplyAsync (() -> "Hallo"); CompletableFuture future2 = CompletableFuture.supplyAsync (() -> "Mooi"); CompletableFuture future3 = CompletableFuture.supplyAsync (() -> "Wereld"); CompletableFuture combinedFuture = CompletableFuture.allOf (future1, future2, future3); // ... combinedFuture.get (); assertTrue (future1.isDone ()); assertTrue (future2.isDone ()); assertTrue (future3.isDone ());

Merk op dat het retourtype van de CompletableFuture.allOf () is een CompletableFuture. De beperking van deze methode is dat het niet de gecombineerde resultaten van alle retourneert Futures. In plaats daarvan moeten we handmatig resultaten halen uit Futures. Gelukkig, CompletableFuture.join () methode en Java 8 Streams API maken het eenvoudig:

String gecombineerd = Stream.of (future1, future2, future3) .map (CompletableFuture :: join) .collect (Collectors.joining ("")); assertEquals ("Hello Beautiful World", gecombineerd);

De CompletableFuture.join () methode is vergelijkbaar met de krijgen methode, maar het genereert een ongecontroleerde uitzondering in het geval dat de Toekomst wordt niet normaal voltooid. Dit maakt het mogelijk om het te gebruiken als methodeverwijzing in het Stream.map () methode.

9. Fouten afhandelen

Voor foutafhandeling in een reeks asynchrone rekenstappen moeten we de gooien vangen idioom op een vergelijkbare manier.

In plaats van een uitzondering op te vangen in een syntactisch blok, kan de CompletableFuture klasse stelt ons in staat om het in een special te behandelen handvat methode. Deze methode ontvangt twee parameters: een resultaat van een berekening (als deze met succes is voltooid) en de gegenereerde uitzondering (als een bepaalde berekeningsstap niet normaal is voltooid).

In het volgende voorbeeld gebruiken we de handvat methode om een ​​standaardwaarde op te geven wanneer de asynchrone berekening van een begroeting is voltooid met een fout omdat er geen naam is opgegeven:

String naam = null; // ... CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> {if (name == null) {throw new RuntimeException ("Computation error!");} Return "Hallo," + naam;})}). handle ((s, t) -> s! = null? s: "Hallo, vreemdeling!"); assertEquals ("Hallo, vreemdeling!", completableFuture.get ());

Stel dat we als alternatief scenario het Toekomst met een waarde, zoals in het eerste voorbeeld, maar hebben ook de mogelijkheid om deze met een uitzondering te voltooien. De compleetUitzonderlijk methode is precies daarvoor bedoeld. De completableFuture.get () methode in het volgende voorbeeld gooit een ExecutionException met een RuntimeException als oorzaak:

CompletableFuture completableFuture = nieuwe CompletableFuture (); // ... completableFuture.completeExceptionally (nieuwe RuntimeException ("Berekening mislukt!")); // ... completableFuture.get (); // ExecutionException

In het bovenstaande voorbeeld hadden we de uitzondering kunnen afhandelen met de handvat methode asynchroon, maar met de krijgen methode kunnen we de meer typische benadering van een synchrone uitzonderingsverwerking gebruiken.

10. Asynchrone methoden

De meeste methoden van de vloeiende API in CompletableFuture klasse hebben twee extra varianten met de Async postfix. Deze methoden zijn meestal bedoeld voor het uitvoeren van een overeenkomstige uitvoeringsstap in een andere thread.

De methoden zonder de Async postfix voert de volgende uitvoeringsfase uit met behulp van een aanroepende thread. In tegenstelling daarmee is de Async methode zonder de Uitvoerder argument voert een stap uit met behulp van de common vork / samenvoegen pool implementatie van Uitvoerder dat toegankelijk is met de ForkJoinPool.commonPool () methode. eindelijk, de Async methode met een Uitvoerder argument voert een stap uit met behulp van de doorgegeven Uitvoerder.

Hier is een aangepast voorbeeld dat het resultaat van een berekening met een Functie voorbeeld. Het enige zichtbare verschil is de thenApplyAsync methode, maar onder de motorkap wordt de toepassing van een functie verpakt in een ForkJoinTask bijvoorbeeld (voor meer informatie over de vork / samenvoegen framework, zie het artikel “Gids voor het Fork / Join Framework in Java”). Dit stelt ons in staat om onze berekeningen nog meer parallel te laten lopen en systeembronnen efficiënter te gebruiken:

CompletableFuture completableFuture = CompletableFuture.supplyAsync (() -> "Hallo"); CompletableFuture future = completableFuture .thenApplyAsync (s -> s + "World"); assertEquals ("Hallo wereld", future.get ());

11. JDK 9 CompletableFuture API

Java 9 verbetert de CompletableFuture API met de volgende wijzigingen:

• Nieuwe fabrieksmethoden toegevoegd
• Ondersteuning voor vertragingen en time-outs
• Verbeterde ondersteuning voor subclassificatie

en nieuwe instantie-API's:

• Executor defaultExecutor ()
• CompletableFuture newIncompleteFuture ()
• Toekomstige kopie ()
• CompletionStage minimalCompletionStage ()
• CompletableFuture completeAsync (leverancier leverancier, uitvoerder uitvoerder)
• CompletableFuture completeAsync (leverancier leverancier)
• CompletableFuture orTimeout (lange time-out, TimeUnit-eenheid)
• CompletableFuture completeOnTimeout (T-waarde, lange time-out, TimeUnit-eenheid)

We hebben nu ook een paar statische hulpprogramma's:

• Executor delayedExecutor (lange vertraging, TimeUnit unit, Executor executor)
• Executor delayedExecutor (lange vertraging, TimeUnit-eenheid)
• CompletionStage completeStage (U-waarde)
• CompletionStage failedStage (Throwable ex)
• CompletableFuture failedFuture (Throwable ex)

Om de time-out aan te pakken, heeft Java 9 tot slot nog twee nieuwe functies geïntroduceerd:

• ofTimeout ()
• completeOnTimeout ()

Hier is het gedetailleerde artikel om verder te lezen: Java 9 CompletableFuture API-verbeteringen.

12. Conclusie

In dit artikel hebben we de methoden en typische gebruiksscenario's van het CompletableFuture klasse.

De broncode voor het artikel is beschikbaar op GitHub.