Razem lepiej: Java i klasa Thread. Część IV — Callable, Future i przyjaciele

Wstęp

W części I omówiliśmy sposób tworzenia wątków. Przypomnijmy jeszcze raz. Wątek jest reprezentowany przez klasę Thread, której run()metoda jest wywoływana. Użyjmy więc internetowego kompilatora Java Tutorialspoint i wykonajmy następujący kod:

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

Czy to jedyna opcja rozpoczęcia zadania w wątku?

java.util.concurrent.Callable

Okazuje się, że java.lang.Runnable ma brata o nazwie java.util.concurrent.Callable , który przyszedł na świat w Javie 1.5. Jakie są różnice? Jeśli przyjrzysz się bliżej Javadoc tego interfejsu, zobaczysz, że w przeciwieństwie do Runnable, nowy interfejs deklaruje call()metodę, która zwraca wynik. Ponadto domyślnie zgłasza wyjątek. Oznacza to, że oszczędza nam konieczności try-catchblokowania sprawdzanych wyjątków. Nieźle, prawda? Teraz mamy nowe zadanie zamiast Runnable:

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

Ale co z tym robimy? Dlaczego potrzebujemy zadania działającego w wątku, który zwraca wynik? Oczywiście w przypadku jakichkolwiek działań wykonanych w przyszłości oczekujemy, że w przyszłości otrzymamy wynik tych działań. I mamy interfejs o odpowiedniej nazwie:java.util.concurrent.Future

java.util.concurrent.Future

Interfejs java.util.concurrent.Future definiuje API do pracy z zadaniami, których wyniki planujemy otrzymać w przyszłości: metody uzyskiwania wyniku i metody sprawdzania statusu. W odniesieniu do Future, jesteśmy zainteresowani jego implementacją w klasie java.util.concurrent.FutureTask . To jest „Zadanie”, które zostanie wykonane w Future. Tym, co czyni tę implementację jeszcze bardziej interesującą, jest to, że implementuje ona również Runnable. Można to uznać za rodzaj adaptera pomiędzy starym modelem pracy z zadaniami na wątkach a nowym (nowym w tym sensie, że pojawił się w Javie 1.5). Oto przykład:

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

Jak widać na przykładzie, używamy getmetody, aby uzyskać wynik z zadania. Notatka:kiedy uzyskasz wynik za pomocą get()metody, wykonanie staje się synchroniczne! Jak myślisz, jaki mechanizm zostanie tutaj zastosowany? To prawda, że ​​​​nie ma bloku synchronizacji. Dlatego nie zobaczymy WAITING w JVisualVM jako monitorlub wait, ale jako znajomą park()metodę (ponieważ LockSupportmechanizm jest używany).

Interfejsy funkcjonalne

Następnie porozmawiamy o klasach z Javy 1.8, więc dobrze byłoby przedstawić krótkie wprowadzenie. Spójrz na następujący 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);
	}
};

Mnóstwo dodatkowego kodu, prawda? Każda z zadeklarowanych klas wykonuje jedną funkcję, ale do jej zdefiniowania używamy kilku dodatkowych kodów pomocniczych. I tak myśleli programiści Javy. W związku z tym wprowadzili zestaw „interfejsów funkcjonalnych” ( @FunctionalInterface) i zdecydowali, że teraz „myśleniem” zajmie się sama Java, pozostawiając nam tylko ważne rzeczy:

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

Zaopatrzenie Supplier. Nie ma parametrów, ale coś zwraca. W ten sposób dostarcza rzeczy. zużywa Consumer. Bierze coś jako dane wejściowe (argument) i coś z tym robi. Argument jest tym, co konsumuje. Wtedy też mamy Function. Pobiera dane wejściowe (argumenty), robi coś i coś zwraca. Widać, że aktywnie używamy leków generycznych. Jeśli nie masz pewności, możesz odświeżyć sobie wiedzę, czytając „ Generiki w Javie: jak używać nawiasów ostrych w praktyce ”.

Ukończona przyszłość

CompletableFutureCzas mijał iw Javie 1.8 pojawiła się nowa klasa o nazwie . Implementuje Futureinterfejs, czyli nasze zadania zostaną wykonane w przyszłości i możemy zadzwonić, get()aby uzyskać wynik. Ale implementuje również CompletionStageinterfejs. Nazwa mówi sama za siebie: to pewien etap pewnego zestawu obliczeń. Krótkie wprowadzenie do tematu można znaleźć w recenzji tutaj: Wprowadzenie do CompletionStage i CompletableFuture. Przejdźmy od razu do rzeczy. Spójrzmy na listę dostępnych metod statycznych, które pomogą nam zacząć: Oto opcje ich użycia:

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

Jeśli wykonamy ten kod, zobaczymy, że utworzenie potoku CompletableFuturewiąże się również z uruchomieniem całego potoku. Dlatego, z pewnym podobieństwem do SteamAPI z Javy 8, tutaj znajdujemy różnicę między tymi podejściami. Na przykład:

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

To jest przykład Stream API Java 8. Jeśli uruchomisz ten kod, zobaczysz, że komunikat „Wykonano” nie zostanie wyświetlony. Innymi słowy, gdy strumień jest tworzony w Javie, strumień nie rozpoczyna się natychmiast. Zamiast tego czeka, aż ktoś będzie chciał uzyskać od niego wartość. Ale CompletableFuturenatychmiast rozpoczyna wykonywanie potoku, nie czekając, aż ktoś poprosi go o wartość. Myślę, że ważne jest, aby to zrozumieć. Tak, mamy CompletableFuture. Jak możemy zrobić potok (lub łańcuch) i jakie mamy mechanizmy? Przypomnij sobie te funkcjonalne interfejsy, o których pisaliśmy wcześniej.

• Mamy a Function, które przyjmuje A i zwraca B. Ma jedną metodę: apply().
• Mamy a Consumer, które przyjmuje A i nic nie zwraca (Void). Ma jedną metodę: accept().
• Mamy Runnable, który działa na wątku i nic nie bierze i nic nie zwraca. Ma jedną metodę: run().

Następną rzeczą do zapamiętania jest to, że CompletableFutureużywa Runnable, Consumers, i Functionsw swojej pracy. W związku z tym zawsze możesz wiedzieć, że możesz wykonać następujące czynności za pomocą 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);
}

Metody thenRun(), thenApply()i thenAccept()mają wersje „Asynchroniczne”. Oznacza to, że etapy te zostaną zakończone na innym wątku. Ten wątek zostanie pobrany ze specjalnej puli — więc nie wiemy z góry, czy będzie to nowy, czy stary wątek. Wszystko zależy od tego, jak intensywne obliczeniowo są zadania. Oprócz tych metod istnieją trzy bardziej interesujące możliwości. Dla jasności wyobraźmy sobie, że mamy pewną usługę, która skądś otrzymuje jakąś wiadomość — a to wymaga czasu:

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

Teraz przyjrzyjmy się innym zdolnościom, które CompletableFuturezapewnia. Możemy połączyć wynik a CompletableFuturez wynikiem innego 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();

Zauważ, że wątki są domyślnie wątkami demonów, więc dla jasności używamy get()oczekiwania na wynik. Nie tylko możemy łączyć CompletableFutures, możemy również zwrócić CompletableFuture:

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

Tutaj chcę zauważyć, że CompletableFuture.completedFuture()metoda została użyta dla zwięzłości. Ta metoda nie tworzy nowego wątku, więc reszta potoku zostanie wykonana w tym samym wątku, w którym completedFuturezostała wywołana. Jest też thenAcceptBoth()metoda. Jest bardzo podobny do accept(), ale jeśli thenAccept()akceptuje Consumer, thenAcceptBoth()akceptuje inny CompletableStage+ BiConsumerjako dane wejściowe, tj. a consumer, które pobiera 2 źródła zamiast jednego. Jest jeszcze jedna interesująca zdolność oferowana przez metody, których nazwa zawiera słowo „Albo”: Metody te akceptują alternatywę CompletableStagei są wykonywane w CompletableStagepierwszej kolejności. Na koniec chcę zakończyć tę recenzję kolejną interesującą funkcją CompletableFuture: obsługą błędów.

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

Ten kod nic nie zrobi, ponieważ będzie wyjątek i nic więcej się nie stanie. Ale odkomentowując stwierdzenie „wyjątkowo”, definiujemy oczekiwane zachowanie. Skoro o tym mowa CompletableFuture, polecam również obejrzenie poniższego filmu:

• Wprowadzenie do CompletableFuture

Moim skromnym zdaniem są to jedne z najbardziej objaśniających filmów w Internecie. Powinni wyjaśnić, jak to wszystko działa, jaki zestaw narzędzi mamy do dyspozycji i dlaczego to wszystko jest potrzebne.

Wniosek

Mamy nadzieję, że teraz jest jasne, w jaki sposób można używać wątków do uzyskiwania obliczeń po ich zakończeniu. Dodatkowy materiał:

• Współbieżność Java (wielowątkowość) — objaśnienie CompletableFuture
• Wprowadzenie do CompletionStage i CompletableFuture
• Współbieżność i ulepszenia w Javie 8: część 2
• Przewodnik po CompletableFuture

Razem lepiej: Java i klasa Thread. Część I — Wątki wykonania Lepiej razem: Java i klasa Thread. Część II — Synchronizacja Lepiej razem: Java i klasa Thread. Część III — Interakcja Lepiej razem: Java i klasa Thread. Część V — Executor, ThreadPool, Fork/Join Better together: Java i klasa Thread. Część VI — Odpalaj!