Samen beter: Java en de klasse Thread. Deel IV — Callable, Future en vrienden

Invoering

In deel I hebben we bekeken hoe threads worden gemaakt. Laten we het ons nog een keer herinneren. Een thread wordt vertegenwoordigd door de klasse Thread, waarvan de run()methode wordt aangeroepen. Laten we dus de Tutorialspoint online Java-compiler gebruiken en de volgende code uitvoeren:

public class HelloWorld {
    
    public static void main(String[] args) {
        Runnable task = () -> {
            System.out.println("Hello World");
        };
        new Thread(task).start();
    }
}

Is dit de enige optie voor het starten van een taak op een thread?

java.util.concurrent.Opvraagbaar

Het blijkt dat java.lang.Runnable een broer heeft genaamd java.util.concurrent.Callable die ter wereld kwam in Java 1.5. Wat zijn de verschillen? Als je de Javadoc voor deze interface goed bekijkt, zien we dat, in tegenstelling tot Runnable, de nieuwe interface een call()methode declareert die een resultaat retourneert. Ook gooit het standaard een uitzondering. Dat wil zeggen, het voorkomt dat we moeten try-catchblokkeren voor aangevinkte uitzonderingen. Niet slecht, toch? Nu hebben we een nieuwe taak in plaats van Runnable:

Callable task = () -> {
	return "Hello, World!";
};

Maar wat doen we ermee? Waarom hebben we een taak nodig die wordt uitgevoerd op een thread die een resultaat retourneert? Vanzelfsprekend verwachten we voor alle acties die in de toekomst worden uitgevoerd, het resultaat van die acties in de toekomst te ontvangen. En we hebben een interface met een bijbehorende naam:java.util.concurrent.Future

java.util.concurrent.Future

De interface java.util.concurrent.Future definieert een API voor het werken met taken waarvan we de resultaten in de toekomst willen ontvangen: methoden om een ​​resultaat te krijgen en methoden om de status te controleren. Met betrekking tot Future, we zijn geïnteresseerd in de implementatie ervan in de klasse java.util.concurrent.FutureTask . Dit is de "taak" die zal worden uitgevoerd in Future. Wat deze implementatie nog interessanter maakt, is dat het ook Runnable implementeert. Je kunt dit beschouwen als een soort adapter tussen het oude model van werken met taken op threads en het nieuwe model (nieuw in de zin dat het verscheen in Java 1.5). Hier is een voorbeeld:

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class HelloWorld {
    
    public static void main(String[] args) throws Exception {
        Callable task = () -> {
            return "Hello, World!";
        };
        FutureTask<String> future = new FutureTask<>(task);
        new Thread(future).start();
        System.out.println(future.get());
    }
}

Zoals je in het voorbeeld kunt zien, gebruiken we de getmethode om het resultaat van de taak te krijgen. Opmerking:wanneer u het resultaat krijgt met behulp van de get()methode, wordt de uitvoering synchroon! Welk mechanisme denk je dat hier zal worden gebruikt? Toegegeven, er is geen synchronisatieblok. Daarom zien we WAITING in JVisualVM niet als een monitorof wait, maar als de vertrouwde park()methode (omdat het LockSupportmechanisme wordt gebruikt).

Functionele interfaces

Vervolgens zullen we het hebben over klassen uit Java 1.8, dus we doen er goed aan een korte introductie te geven. Kijk naar de volgende code:

Supplier<String> supplier = new Supplier<String>() {
	@Override
	public String get() {
		return "String";
	}
};
Consumer<String> consumer = new Consumer<String>() {
	@Override
	public void accept(String s) {
		System.out.println(s);
	}
};
Function<String, Integer> converter = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return Integer.valueOf(s);
	}
};

Heel veel extra code, zou je niet zeggen? Elk van de gedeclareerde klassen voert één functie uit, maar we gebruiken een heleboel extra ondersteunende code om deze te definiëren. En zo dachten Java-ontwikkelaars. Dienovereenkomstig introduceerden ze een reeks "functionele interfaces" ( @FunctionalInterface) en besloten dat Java nu zelf het "denkwerk" zou doen, waardoor alleen de belangrijke dingen voor ons overbleven om ons zorgen over te maken:

Supplier<String> supplier = () -> "String";
Consumer<String> consumer = s -> System.out.println(s);
Function<String, Integer> converter = s -> Integer.valueOf(s);

Een Supplierbenodigdheden. Het heeft geen parameters, maar het geeft iets terug. Dit is hoe het dingen levert. A Consumerverbruikt. Het neemt iets als invoer (een argument) en doet er iets mee. Het argument is wat het verbruikt. Dan hebben we ook Function. Het neemt input (argumenten), doet iets en geeft iets terug. U kunt zien dat we actief generieke geneesmiddelen gebruiken. Als u het niet zeker weet, kunt u een opfriscursus krijgen door " Generics in Java: how to use punthaken in de praktijk " te lezen.

VoltooibareToekomst

De tijd verstreek en een nieuwe klasse genaamd CompletableFutureverscheen in Java 1.8. Het implementeert de Futureinterface, dwz onze taken zullen in de toekomst worden voltooid en we kunnen bellen get()om het resultaat te krijgen. Maar het implementeert ook de CompletionStageinterface. De naam zegt het al: dit is een bepaalde fase van een reeks berekeningen. Een korte inleiding tot het onderwerp is te vinden in de recensie hier: Inleiding tot CompletionStage en CompletableFuture. Laten we meteen ter zake komen. Laten we eens kijken naar de lijst met beschikbare statische methoden waarmee we aan de slag kunnen: Hier zijn opties om ze te gebruiken:

import java.util.concurrent.CompletableFuture;
public class App {
    public static void main(String[] args) throws Exception {
        // A CompletableFuture that already contains a Result
        CompletableFuture<String> completed;
        completed = CompletableFuture.completedFuture("Just a value");
        // A CompletableFuture that runs a new thread from Runnable. That's why it's Void
        CompletableFuture<Void> voidCompletableFuture;
        voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("run " + Thread.currentThread().getName());
        });
        // A CompletableFuture that starts a new thread whose result we'll get from a Supplier 
        CompletableFuture<String> supplier;
        supplier = CompletableFuture.supplyAsync(() -> {
            System.out.println("supply " + Thread.currentThread().getName());
            return "Value";
        });
    }
}

Als we deze code uitvoeren, zien we dat het maken van een CompletableFutureook het starten van een hele pijplijn inhoudt. Daarom, met een zekere gelijkenis met de SteamAPI van Java8, vinden we hier het verschil tussen deze benaderingen. Bijvoorbeeld:

List<String> array = Arrays.asList("one", "two");
Stream<String> stringStream = array.stream().map(value -> {
	System.out.println("Executed");
	return value.toUpperCase();
});

Dit is een voorbeeld van de Stream API van Java 8. Als u deze code uitvoert, ziet u dat "Uitgevoerd" niet wordt weergegeven. Met andere woorden, wanneer een stream in Java wordt aangemaakt, start de stream niet meteen. In plaats daarvan wacht het tot iemand er een waarde uit wil halen. Maar CompletableFuturebegint de pijplijn onmiddellijk uit te voeren, zonder te wachten tot iemand hem om een ​​waarde vraagt. Ik denk dat dit belangrijk is om te begrijpen. Zo, we hebben een CompletableFuture. Hoe kunnen we een pijpleiding (of ketting) maken en welke mechanismen hebben we? Denk aan die functionele interfaces waarover we eerder schreven.

• We hebben a Functiondie een A neemt en een B retourneert. Het heeft een enkele methode: apply().
• We hebben a Consumerdie een A neemt en niets retourneert (Void). Het heeft een enkele methode: accept().
• We hebben Runnable, die op de thread draait, en neemt niets en retourneert niets. Het heeft een enkele methode: run().

Het volgende dat u moet onthouden, is dat , , en in zijn werk CompletableFuturewordt gebruikt . Zo weet u altijd dat u het volgende kunt doen met : RunnableConsumersFunctionsCompletableFuture

public static void main(String[] args) throws Exception {
        AtomicLong longValue = new AtomicLong(0);
        Runnable task = () -> longValue.set(new Date().getTime());
        Function<Long, Date> dateConverter = (longvalue) -> new Date(longvalue);
        Consumer<Date> printer = date -> {
            System.out.println(date);
            System.out.flush();
        };
        // CompletableFuture computation
        CompletableFuture.runAsync(task)
                         .thenApply((v) -> longValue.get())
                         .thenApply(dateConverter)
                         .thenAccept(printer);
}

De methoden thenRun(), thenApply(), en thenAccept()hebben "Asynchrone" versies. Dit betekent dat deze fasen op een andere thread worden voltooid. Deze thread komt uit een speciale pool — we weten dus niet van tevoren of het een nieuwe of oude thread wordt. Het hangt allemaal af van hoe rekenintensief de taken zijn. Naast deze methoden zijn er nog drie interessante mogelijkheden. Laten we ons voor de duidelijkheid voorstellen dat we een bepaalde dienst hebben die ergens een bericht van ontvangt — en dit kost tijd:

public static class NewsService {
	public static String getMessage() {
		try {
			Thread.currentThread().sleep(3000);
			return "Message";
		} catch (InterruptedException e) {
			throw new IllegalStateException(e);
		}
	}
}

Laten we nu eens kijken naar andere mogelijkheden die CompletableFuturebieden. We kunnen het resultaat van a combineren CompletableFuturemet het resultaat van een ander CompletableFuture:

Supplier newsSupplier = () -> NewsService.getMessage();
        
CompletableFuture<String> reader = CompletableFuture.supplyAsync(newsSupplier);
CompletableFuture.completedFuture("!!")
				 .thenCombine(reader, (a, b) -> b + a)
				 .thenAccept(result -> System.out.println(result))
				 .get();

Merk op dat threads standaard daemon-threads zijn, dus voor de duidelijkheid get()wachten we op het resultaat. We kunnen niet alleen combineren CompletableFutures, we kunnen ook een CompletableFuture:

CompletableFuture.completedFuture(2L)
				.thenCompose((val) -> CompletableFuture.completedFuture(val + 2))
                               .thenAccept(result -> System.out.println(result));

Hier wil ik opmerken dat de CompletableFuture.completedFuture()methode werd gebruikt voor beknoptheid. Deze methode maakt geen nieuwe thread aan, dus de rest van de pijplijn wordt uitgevoerd op dezelfde thread waar deze completedFuturewerd aangeroepen. Er is ook een thenAcceptBoth()methode. Het lijkt erg op accept(), maar als thenAccept()a wordt geaccepteerd Consumer, wordt een andere + als invoer thenAcceptBoth()geaccepteerd , dwz a die 2 bronnen nodig heeft in plaats van één. Er is nog een andere interessante mogelijkheid die methoden bieden waarvan de naam het woord "Ofwel" bevat: deze methoden accepteren een alternatief en worden uitgevoerd op de manier die als eerste wordt uitgevoerd. Ten slotte wil ik deze recensie beëindigen met een ander interessant kenmerk van : foutafhandeling. CompletableStageBiConsumerconsumerCompletableStageCompletableStageCompletableFuture

CompletableFuture.completedFuture(2L)
				 .thenApply((a) -> {
					throw new IllegalStateException("error");
				 }).thenApply((a) -> 3L)
				 //.exceptionally(ex -> 0L)
				 .thenAccept(val -> System.out.println(val));

Deze code zal niets doen, omdat er een uitzondering zal zijn en er verder niets zal gebeuren. Maar door de opmerking "uitzonderlijk" weg te laten, definiëren we het verwachte gedrag. Nu we het toch over hebben CompletableFuture, ik raad je ook aan om de volgende video te bekijken:

• Inleiding tot CompletableFuture

Naar mijn bescheiden mening behoren dit tot de meest verklarende video's op internet. Ze moeten duidelijk maken hoe dit allemaal werkt, welke toolkit we beschikbaar hebben en waarom dit allemaal nodig is.

Conclusie

Hopelijk is het nu duidelijk hoe je threads kunt gebruiken om berekeningen te krijgen nadat ze zijn voltooid. Aanvullend materiaal:

• Java-concurrency (multithreading) — CompletableFuture uitgelegd
• Inleiding tot CompletionStage en CompletableFuture
• Gelijktijdigheid en verbeteringen in Java 8: deel 2
• Gids voor CompletableFuture

Samen beter: Java en de klasse Thread. Deel I — Uitvoeringsthreads Beter samen: Java en de klasse Thread. Deel II — Synchronisatie Samen beter: Java en de klasse Thread. Deel III — Interactie Samen beter: Java en de klasse Thread. Deel V — Uitvoerder, ThreadPool, Fork/Join Beter samen: Java en de Thread-klasse. Deel VI — Vuur weg!