Bättre tillsammans: Java och trådklassen. Del IV — Callable, Future och vänner

Introduktion

I del I gick vi igenom hur trådar skapas. Låt oss minnas en gång till. En tråd representeras av klassen Thread, vars run()metod anropas. Så låt oss använda Tutorialspoints online Java-kompilator och köra följande kod:

public class HelloWorld {

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

Är detta det enda alternativet för att starta en uppgift i en tråd?

java.util.concurrent.Callable

Det visar sig att java.lang.Runnable har en bror som heter java.util.concurrent.Callable som kom till världen i Java 1.5. Vilka är skillnaderna? Om du tittar noga på Javadoc för detta gränssnitt ser vi att, till skillnad från , Runnabledeklarerar det nya gränssnittet en call()metod som returnerar ett resultat. Dessutom kastar det undantag som standard. Det vill säga, det besparar oss från att behöva try-catchblockera för kontrollerade undantag. Inte illa, eller hur? Nu har vi en ny uppgift istället för Runnable:

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

Men vad gör vi med det? Varför behöver vi en uppgift som körs på en tråd som returnerar ett resultat? För alla åtgärder som utförs i framtiden förväntar vi oss naturligtvis att få resultatet av dessa åtgärder i framtiden. Och vi har ett gränssnitt med ett motsvarande namn:java.util.concurrent.Future

java.util.concurrent.Future

Gränssnittet java.util.concurrent.Future definierar ett API för att arbeta med uppgifter vars resultat vi planerar att få i framtiden: metoder för att få ett resultat och metoder för att kontrollera status. När det gäller Future, är vi intresserade av dess implementering i klassen java.util.concurrent.FutureTask . Detta är "uppgiften" som kommer att utföras i Future. Det som gör den här implementeringen ännu mer intressant är att den även implementerar Runnable. Du kan betrakta detta som ett slags adapter mellan den gamla modellen att arbeta med uppgifter på trådar och den nya modellen (ny i den meningen att den dök upp i Java 1.5). Här är ett exempel:

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 i exemplet använder vi getmetoden för att få resultatet av uppgiften. Notera:när du får resultatet med get()metoden blir exekveringen synkron! Vilken mekanism tror du kommer att användas här? Det är sant att det inte finns något synkroniseringsblock. Det är därför vi inte kommer att se WAITING i JVisualVM som en monitoreller wait, utan som den välbekanta park()metoden (eftersom LockSupportmekanismen används).

Funktionella gränssnitt

Därefter kommer vi att prata om klasser från Java 1.8, så vi gör klokt i att ge en kort introduktion. Titta på följande kod:

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

Massor och massor av extra kod, skulle du inte säga? Var och en av de deklarerade klasserna utför en funktion, men vi använder en massa extra stödkod för att definiera den. Och så här tänkte Java-utvecklare. Följaktligen introducerade de en uppsättning "funktionella gränssnitt" ( @FunctionalInterface) och beslutade att nu Java själv skulle göra "tänket", vilket bara lämnar de viktiga sakerna för oss att oroa oss för:

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

A Supplierförnödenheter. Den har inga parametrar, men den returnerar något. Det är så det ger saker. A Consumerförbrukar. Den tar något som en input (ett argument) och gör något med det. Argumentet är vad det förbrukar. Sen har vi också Function. Det tar input (argument), gör något och returnerar något. Du kan se att vi aktivt använder generika. Om du är osäker kan du få en repetition genom att läsa " Generics in Java: how to use angled brackets in practice " .

CompletableFuture

Tiden gick och en ny klass som heter CompletableFuturedök upp i Java 1.8. Den implementerar Futuregränssnittet, dvs våra uppgifter kommer att slutföras i framtiden, och vi kan ringa get()för att få resultatet. Men det implementerar också CompletionStagegränssnittet. Namnet säger allt: detta är ett visst skede av någon uppsättning beräkningar. En kort introduktion till ämnet finns i recensionen här: Introduktion till CompletionStage och CompletableFuture. Låt oss gå rätt till saken. Låt oss titta på listan över tillgängliga statiska metoder som hjälper oss att komma igång: Här är alternativen för att använda 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";
        });
    }
}

Om vi ​​kör den här koden kommer vi att se att skapande av en CompletableFutureockså innebär att en hel pipeline lanseras. Därför, med en viss likhet med SteamAPI från Java8, är det här vi hittar skillnaden mellan dessa tillvägagångssätt. Till exempel:

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

Detta är ett exempel på Java 8:s Stream API. Om du kör den här koden kommer du att se att "Executed" inte kommer att visas. Med andra ord, när en stream skapas i Java startar inte streamen omedelbart. Istället väntar den på att någon vill ha ett värde av den. Men CompletableFuturebörjar exekvera pipelinen omedelbart, utan att vänta på att någon ska fråga den om ett värde. Jag tror att detta är viktigt att förstå. S o, vi har en CompletableFuture. Hur kan vi göra en pipeline (eller kedja) och vilka mekanismer har vi? Kom ihåg de funktionella gränssnitten som vi skrev om tidigare.

• Vi har ett Functionsom tar ett A och returnerar ett B. Det har en enda metod: apply().
• Vi har ett Consumersom tar ett A och inte returnerar något (Void). Den har en enda metod: accept().
• Vi har Runnable, som körs på tråden, och tar ingenting och returnerar ingenting. Den har en enda metod: run().

Nästa sak att komma ihåg är att CompletableFutureanvänder , Runnable, Consumersoch Functionsi sitt arbete. Följaktligen kan du alltid veta att du kan göra följande 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);
}

Metoderna thenRun(), thenApply(), och thenAccept()har "Async"-versioner. Detta innebär att dessa steg kommer att slutföras på en annan tråd. Den här tråden kommer att tas från en speciell pool — så vi vet inte i förväg om det blir en ny eller gammal tråd. Allt beror på hur beräkningsintensiva uppgifterna är. Utöver dessa metoder finns det ytterligare tre intressanta möjligheter. För tydlighetens skull, låt oss föreställa oss att vi har en viss tjänst som tar emot något slags meddelande någonstans ifrån - och detta tar tid:

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

Låt oss nu ta en titt på andra förmågor som CompletableFutureger. Vi kan kombinera resultatet av ett CompletableFuturemed resultatet av ett annat 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();

Observera att trådar är demontrådar som standard, så för tydlighetens skull använder vi get()för att vänta på resultatet. Vi kan inte bara kombinera CompletableFutures, vi kan också returnera en CompletableFuture:

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

Här vill jag notera att CompletableFuture.completedFuture()metoden användes för korthets skull. Denna metod skapar inte en ny tråd, så resten av pipelinen kommer att exekveras på samma tråd som completedFutureanropades. Det finns också en thenAcceptBoth()metod. Det är väldigt likt accept(), men om thenAccept()accepterar en Consumer, thenAcceptBoth()accepterar en annan CompletableStage+ BiConsumersom indata, dvs en consumersom tar 2 källor istället för en. Det finns en annan intressant förmåga som erbjuds av metoder vars namn innehåller ordet "Antingen": Dessa metoder accepterar ett alternativ CompletableStageoch exekveras på den CompletableStagesom exekveras först. Till sist vill jag avsluta den här recensionen med en annan intressant egenskap CompletableFuture: felhantering.

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

Den här koden kommer inte att göra något, eftersom det kommer att finnas ett undantag och inget annat kommer att hända. Men genom att avkommentera "exceptionellt" uttalandet definierar vi det förväntade beteendet. På tal om CompletableFuture, jag rekommenderar dig också att titta på följande video:

• Introduktion till CompletableFuture

Enligt min ödmjuka åsikt är dessa bland de mest förklarande videorna på Internet. De bör göra det klart hur allt detta fungerar, vilken verktygslåda vi har tillgänglig och varför allt detta behövs.

Slutsats

Förhoppningsvis är det nu klart hur du kan använda trådar för att få beräkningar efter att de är klara. Ytterligare material:

• Java Concurrency (Multithreading) — CompletableFuture Explained
• Introduktion till CompletionStage och CompletableFuture
• Samtidighet och förbättringar i Java 8: Del 2
• Guide till CompletableFuture

Bättre tillsammans: Java och trådklassen. Del I — Trådar av utförande Bättre tillsammans: Java och klassen Thread. Del II — Synkronisering Bättre tillsammans: Java och klassen Thread. Del III — Interaktion Bättre tillsammans: Java och klassen Thread. Del V — Executor, ThreadPool, Fork/Join Better tillsammans: Java och Thread-klassen. Del VI — Skjut loss!