Bedre sammen: Java og Tråd-klassen. Del IV — Callable, Future og venner

Introduktion

I del I gennemgik vi, hvordan tråde oprettes. Lad os huske en gang mere. En tråd er repræsenteret af klassen Thread, hvis run()metode bliver kaldt. Så lad os bruge Tutorialspoint online Java compiler og udføre følgende kode:

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

Er dette den eneste mulighed for at starte en opgave i en tråd?

java.util.concurrent.Callable

Det viser sig, at java.lang.Runnable har en bror ved navn java.util.concurrent.Callable , som kom til verden i Java 1.5. Hvad er forskellene? Hvis du ser nærmere på Javadoc for denne grænseflade, ser vi, at i modsætning til Runnable, erklærer den nye grænseflade en call()metode, der returnerer et resultat. Det kaster også Undtagelse som standard. Det vil sige, at det sparer os for at skulle try-catchblokere for kontrollerede undtagelser. Ikke dårligt, vel? Nu har vi en ny opgave i stedet for Runnable:

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

Men hvad skal vi med det? Hvorfor har vi brug for en opgave, der kører på en tråd, der returnerer et resultat? For enhver handling, der udføres i fremtiden, forventer vi naturligvis at modtage resultatet af disse handlinger i fremtiden. Og vi har en grænseflade med et tilsvarende navn:java.util.concurrent.Future

java.util.samtidig.Fremtid

Java.util.concurrent.Future - grænsefladen definerer en API til at arbejde med opgaver, hvis resultater vi planlægger at modtage i fremtiden: metoder til at få et resultat og metoder til at kontrollere status. Med hensyn til Future, er vi interesserede i dens implementering i klassen java.util.concurrent.FutureTask . Dette er "Opgaven", der vil blive udført i Future. Det, der gør denne implementering endnu mere interessant, er, at den også implementerer Runnable. Du kan betragte dette som en slags adapter mellem den gamle model for at arbejde med opgaver på tråde og den nye model (ny i den forstand, at den dukkede op i Java 1.5). Her er et eksempel:

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());
    }
}

Som du kan se af eksemplet, bruger vi getmetoden til at få resultatet fra opgaven. Bemærk:når du får resultatet ved hjælp af get()metoden, bliver eksekveringen synkron! Hvilken mekanisme tror du vil blive brugt her? Sandt nok er der ingen synkroniseringsblok. Derfor vil vi ikke se WAITING i JVisualVM som en monitoreller wait, men som den velkendte park()metode (fordi LockSupportmekanismen bliver brugt).

Funktionelle grænseflader

Dernæst vil vi tale om klasser fra Java 1.8, så vi gør klogt i at give en kort introduktion. Se på følgende kode:

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);
	}
};

Masser og masser af ekstra kode, ville du ikke sige? Hver af de erklærede klasser udfører en funktion, men vi bruger en masse ekstra understøttende kode til at definere den. Og sådan tænkte Java-udviklere. I overensstemmelse hermed introducerede de et sæt "funktionelle grænseflader" ( @FunctionalInterface) og besluttede, at nu ville Java selv gøre "tænkningen", hvilket kun efterlod de vigtige ting for os at bekymre os om:

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

A Supplierforsyninger. Den har ingen parametre, men den returnerer noget. Sådan leverer den tingene. A Consumerforbruger. Den tager noget som input (et argument) og gør noget med det. Argumentet er, hvad det forbruger. Så har vi også Function. Det tager input (argumenter), gør noget og returnerer noget. Du kan se, at vi aktivt bruger generika. Hvis du er usikker, kan du få en genopfriskning ved at læse " Generics in Java: how to use angled brackets in practice ".

CompletableFuture

Tiden gik, og en ny klasse kaldet CompletableFuturedukkede op i Java 1.8. Den implementerer Futuregrænsefladen, dvs. vores opgaver vil blive løst i fremtiden, og vi kan ringe get()for at få resultatet. Men det implementerer også CompletionStagegrænsefladen. Navnet siger det hele: dette er et bestemt trin i et sæt beregninger. En kort introduktion til emnet kan findes i anmeldelsen her: Introduktion til CompletionStage og CompletableFuture. Lad os komme lige til sagen. Lad os se på listen over tilgængelige statiske metoder, der vil hjælpe os i gang: Her er muligheder for at bruge dem:

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";
        });
    }
}

Hvis vi udfører denne kode, vil vi se, at oprettelse af en CompletableFutureogså involverer lancering af en hel pipeline. Derfor, med en vis lighed med SteamAPI'en fra Java8, er det her, vi finder forskellen mellem disse tilgange. For eksempel:

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

Dette er et eksempel på Java 8's Stream API. Hvis du kører denne kode, vil du se, at "Udført" ikke vil blive vist. Med andre ord, når en stream oprettes i Java, starter streamingen ikke med det samme. I stedet venter den på, at nogen vil have en værdi af den. Men CompletableFuturebegynder at eksekvere pipelinen med det samme uden at vente på, at nogen spørger den om en værdi. Jeg tror, ​​det er vigtigt at forstå. Så vi har en CompletableFuture. Hvordan kan vi lave en pipeline (eller kæde) og hvilke mekanismer har vi? Husk de funktionelle grænseflader, som vi skrev om tidligere.

• Vi har en Function, der tager et A og returnerer et B. Det har en enkelt metode: apply().
• Vi har en Consumer, der tager et A og ikke returnerer noget (Void). Det har en enkelt metode: accept().
• Vi har Runnable, som kører på tråden, og tager intet og returnerer intet. Det har en enkelt metode: run().

Den næste ting at huske er, at CompletableFuturebruger Runnable, Consumers, og Functionsi sit arbejde. Derfor kan du altid vide, at du kan gøre følgende med CompletableFuture:

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);
}

Metoderne thenRun(), thenApply(), og thenAccept()har "Async"-versioner. Det betyder, at disse faser vil blive gennemført på en anden tråd. Denne tråd vil blive taget fra en speciel pulje — så vi ved ikke på forhånd, om det bliver en ny eller gammel tråd. Det hele afhænger af, hvor beregningstunge opgaverne er. Ud over disse metoder er der yderligere tre interessante muligheder. For klarhedens skyld, lad os forestille os, at vi har en bestemt tjeneste, der modtager en form for besked et eller andet sted fra - og det tager tid:

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

Lad os nu tage et kig på andre evner, der CompletableFuturegiver. Vi kan kombinere resultatet af et CompletableFuturemed resultatet af et andet 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();

Bemærk, at tråde er dæmontråde som standard, så for klarhedens skyld bruger vi get()at vente på resultatet. Ikke alene kan vi kombinere CompletableFutures, vi kan også returnere en CompletableFuture:

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

Her vil jeg bemærke, at CompletableFuture.completedFuture()metoden blev brugt for kortheds skyld. Denne metode opretter ikke en ny tråd, så resten af ​​pipelinen vil blive udført på den samme tråd, hvor den completedFutureblev kaldt. Der er også en thenAcceptBoth()metode. Det minder meget om accept(), men hvis thenAccept()accepterer et Consumer, thenAcceptBoth()accepterer det et andet CompletableStage+ BiConsumersom input, altså en consumerder tager 2 kilder i stedet for én. Der er en anden interessant evne, der tilbydes af metoder, hvis navn inkluderer ordet "Enten": Disse metoder accepterer et alternativ CompletableStageog udføres på den CompletableStage, der udføres først. Til sidst vil jeg afslutte denne anmeldelse med et andet interessant træk ved CompletableFuture: fejlhåndtering.

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

Denne kode vil ikke gøre noget, fordi der vil være en undtagelse, og intet andet vil ske. Men ved at afkommentere "undtagelsesvis" udsagnet, definerer vi den forventede adfærd. Apropos CompletableFuture, så anbefaler jeg dig også at se følgende video:

• Introduktion til CompletableFuture

Efter min ydmyge mening er disse blandt de mest forklarende videoer på internettet. De bør gøre det klart, hvordan det hele fungerer, hvilket værktøjssæt vi har til rådighed, og hvorfor alt dette er nødvendigt.

Konklusion

Forhåbentlig er det nu klart, hvordan du kan bruge tråde til at få beregninger, når de er afsluttet. Yderligere materiale:

• Java Concurrency (Multithreading) — CompletableFuture Explained
• Introduktion til CompletionStage og CompletableFuture
• Samtidighed og forbedringer i Java 8: Del 2
• Guide til CompletableFuture

Bedre sammen: Java og Tråd-klassen. Del I — Udførelsestråde Bedre sammen: Java og trådklassen. Del II — Synkronisering Bedre sammen: Java og Thread-klassen. Del III — Interaktion Bedre sammen: Java og Thread-klassen. Del V — Executor, ThreadPool, Fork/Join Better sammen: Java og Thread-klassen. Del VI - Fyr væk!