Bedre sammen: Java og Thread-klassen. Del IV - Callable, Future og venner

Introduksjon

I del I gjennomgikk vi hvordan tråder opprettes. La oss huske en gang til. En tråd er representert av Thread-klassen, hvis run()metode blir kalt. Så la oss bruke Tutorialspoint online Java-kompilatoren og kjø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 det eneste alternativet for å starte en oppgave i en tråd?

java.util.concurrent.Callable

Det viser seg at java.lang.Runnable har en bror som heter java.util.concurrent.Callable som kom til verden i Java 1.5. Hva er forskjellene? Hvis du ser nøye på Javadoc for dette grensesnittet, ser vi at i motsetning til Runnable, erklærer det nye grensesnittet en call()metode som returnerer et resultat. Dessuten kaster det unntak som standard. Det vil si at det sparer oss for å måtte try-catchblokkere for sjekkede unntak. Ikke verst, ikke sant? Nå har vi en ny oppgave i stedet for Runnable:

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

Men hva gjør vi med det? Hvorfor trenger vi en oppgave som kjører på en tråd som returnerer et resultat? For alle handlinger som utføres i fremtiden, forventer vi selvsagt å motta resultatet av disse handlingene i fremtiden. Og vi har et grensesnitt med et tilsvarende navn:java.util.concurrent.Future

java.util.samtidig.Fremtid

Grensesnittet java.util.concurrent.Future definerer et API for å jobbe med oppgaver som vi planlegger å motta resultater av i fremtiden: metoder for å få et resultat, og metoder for å sjekke status. Når det gjelder Future, er vi interessert i implementeringen i klassen java.util.concurrent.FutureTask . Dette er "Oppgaven" som vil bli utført i Future. Det som gjør denne implementeringen enda mer interessant er at den også implementerer Runnable. Du kan betrakte dette som en slags adapter mellom den gamle modellen for å jobbe med oppgaver på tråder og den nye modellen (ny i den forstand at den dukket opp 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 ser av eksempelet bruker vi getmetoden for å få resultatet fra oppgaven. Merk:når du får resultatet ved hjelp av get()metoden, blir utførelse synkron! Hvilken mekanisme tror du vil bli brukt her? Riktignok er det ingen synkroniseringsblokk. Derfor vil vi ikke se WAITING i JVisualVM som en monitoreller wait, men som den kjente park()metoden (fordi LockSupportmekanismen brukes).

Funksjonelle grensesnitt

Deretter skal vi snakke om klasser fra Java 1.8, så vi gjør klokt i å gi en kort introduksjon. 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);
	}
};

Masse og masse ekstra kode, vil du ikke si? Hver av de deklarerte klassene utfører én funksjon, men vi bruker en haug med ekstra støttekode for å definere den. Og dette er hvordan Java-utviklere tenkte. Følgelig introduserte de et sett med "funksjonelle grensesnitt" ( @FunctionalInterface) og bestemte at nå ville Java selv gjøre "tenkningen", og la bare de viktige tingene for oss å bekymre oss for:

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 parametere, men den returnerer noe. Slik leverer det ting. A Consumerforbruker. Den tar noe som input (et argument) og gjør noe med det. Argumentet er hva det forbruker. Da har vi også Function. Den tar input (argumenter), gjør noe og returnerer noe. Du kan se at vi aktivt bruker generika. Hvis du er usikker, kan du få en oppfriskning ved å lese " Generics in Java: how to use angled brackets in practice ".

CompletableFuture

Tiden gikk og en ny klasse kalt CompletableFuturedukket opp i Java 1.8. Den implementerer Futuregrensesnittet, det vil si at oppgavene våre vil bli utført i fremtiden, og vi kan ringe get()for å få resultatet. Men den implementerer også CompletionStagegrensesnittet. Navnet sier alt: dette er et visst stadium i et sett med beregninger. En kort introduksjon til emnet finner du i anmeldelsen her: Introduction to CompletionStage og CompletableFuture. La oss komme rett til poenget. La oss se på listen over tilgjengelige statiske metoder som vil hjelpe oss å komme i gang: Her er alternativer for å bruke 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 kjører denne koden, vil vi se at å lage en CompletableFutureogså innebærer å lansere en hel pipeline. Derfor, med en viss likhet med SteamAPI fra Java8, er det her vi finner forskjellen mellom disse tilnærmingene. 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 8s Stream API. Hvis du kjører denne koden, vil du se at «Utført» ikke vises. Med andre ord, når en strøm er opprettet i Java, starter ikke strømmen umiddelbart. I stedet venter den på at noen vil ha en verdi fra den. Men CompletableFuturebegynner å utføre rørledningen umiddelbart, uten å vente på at noen skal spørre den om en verdi. Jeg tror dette er viktig å forstå. S o, vi har en CompletableFuture. Hvordan kan vi lage en rørledning (eller kjede) og hvilke mekanismer har vi? Husk de funksjonelle grensesnittene som vi skrev om tidligere.

• Vi har en Functionsom tar en A og returnerer en B. Den har en enkelt metode: apply().
• Vi har en Consumersom tar en A og ikke returnerer noe (Void). Den har en enkelt metode: accept().
• Vi har Runnable, som går på tråden, og tar ingenting og returnerer ingenting. Den har en enkelt metode: run().

Den neste tingen å huske er at CompletableFuturebruker Runnable, Consumers, og Functionsi sitt arbeid. Følgelig kan du alltid vite at du kan gjø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);
}

Metodene thenRun(), thenApply(), og thenAccept()har "Async"-versjoner. Dette betyr at disse stadiene vil bli fullført på en annen tråd. Denne tråden vil bli hentet fra en spesiell gruppe — så vi vet ikke på forhånd om det blir en ny eller gammel tråd. Alt avhenger av hvor beregningsintensive oppgavene er. I tillegg til disse metodene er det tre flere interessante muligheter. For klarhetens skyld, la oss forestille oss at vi har en bestemt tjeneste som mottar en slags melding fra et sted – og dette tar tid:

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

La oss nå ta en titt på andre evner som CompletableFuturegir. Vi kan kombinere resultatet av en CompletableFuturemed resultatet av en annen 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 at tråder er demon-tråder som standard, så for klarhetens skyld bruker vi get()å vente på resultatet. Ikke bare 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 bemerke at CompletableFuture.completedFuture()metoden ble brukt for korthets skyld. Denne metoden oppretter ikke en ny tråd, så resten av rørledningen vil bli utført på samme tråd som completedFutureble kalt. Det finnes også en thenAcceptBoth()metode. Den er veldig lik accept(), men hvis thenAccept()aksepterer en Consumer, thenAcceptBoth()aksepterer en annen CompletableStage+ BiConsumersom input, dvs. en consumersom tar 2 kilder i stedet for én. Det er en annen interessant evne som tilbys av metoder hvis navn inkluderer ordet "Enten": Disse metodene aksepterer et alternativ CompletableStageog blir utført på den CompletableStagesom skal utføres først. Til slutt vil jeg avslutte denne anmeldelsen med en annen interessant funksjon ved CompletableFuture: feilhåndtering.

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

Denne koden vil ikke gjøre noe, fordi det vil være et unntak og ingenting annet vil skje. Men ved å avkommentere "eksepsjonelt" utsagnet, definerer vi forventet atferd. Apropos CompletableFuture, jeg anbefaler deg også å se følgende video:

• Introduksjon til CompletableFuture

Etter min ydmyke mening er disse blant de mest forklarende videoene på Internett. De bør gjøre det klart hvordan alt dette fungerer, hvilket verktøysett vi har tilgjengelig, og hvorfor alt dette er nødvendig.

Konklusjon

Forhåpentligvis er det nå klart hvordan du kan bruke tråder for å få beregninger etter at de er fullført. Tilleggsmateriale:

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

Bedre sammen: Java og Thread-klassen. Del I — Tråder av utførelse Bedre sammen: Java og trådklassen. Del II — Synkronisering Bedre sammen: Java og Thread-klassen. Del III — Interaksjon Bedre sammen: Java og Thread-klassen. Del V — Executor, ThreadPool, Fork/Join Better together: Java og Thread-klassen. Del VI – Fyr vekk!