По-добре заедно: Java и клас Thread. Част IV — Извикваеми, Бъдещи и приятели

Въведение

В част I прегледахме How се създават нишки. Да си припомним още веднъж. Нишката е представена от класа Thread, чийто run()метод се извиква. Така че нека използваме онлайн компилатора на Java на Tutorialspoint и изпълним следния code:

public class HelloWorld {

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

Това ли е единствената опция за стартиране на задача в нишка?

java.util.concurrent.Callable

Оказва се, че java.lang.Runnable има брат, наречен java.util.concurrent.Callable , който се появи на света в Java 1.5. Какви са разликите? Ако погледнете внимателно Javadoc за този интерфейс, виждаме, че за разлика от Runnable, новият интерфейс декларира call()метод, който връща резултат. Освен това хвърля изключение по подразбиране. Тоест, спестява ни от необходимостта да try-catchблокираме за проверени изключения. Не е лошо, нали? Сега имаме нова задача instead of Runnable:

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

Но Howво да правим с него? Защо се нуждаем от задача, изпълнявана в нишка, която връща резултат? Очевидно за всички действия, извършени в бъдеще, очакваме да получим резултата от тези действия в бъдеще. И имаме интерфейс със съответно име:java.util.concurrent.Future

java.util.concurrent.Future

Интерфейсът java.util.concurrent.Future дефинира API за работа със задачи, чиито резултати планираме да получим в бъдеще: методи за получаване на резултат и методи за проверка на състоянието. По отношение на Future, ние се интересуваме от неговата реализация в класа java.util.concurrent.FutureTask . Това е „Задачата“, която ще бъде изпълнена в Future. Това, което прави тази реализация още по-интересна е, че тя също така имплементира Runnable. Можете да считате това за вид адаптер между стария модел на работа със задачи върху нишки и новия модел (нов в смисъл, че се появи в Java 1.5). Ето един пример:

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

Както можете да видите от примера, използваме getметода, за да получим резултата от задачата. Забележка:когато получите резултата с помощта на get()метода, изпълнението става синхронно! Какъв механизъм смятате, че ще се използва тук? Вярно е, че няма блок за синхронизация. Ето защо няма да видим WAITING в JVisualVM като monitoror wait, а като познатия park()метод (тъй като LockSupportмеханизмът се използва).

Функционални интерфейси

След това ще говорим за класове от Java 1.8, така че би било добре да предоставим кратко въведение. Вижте следния 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);
	}
};

Много и много допълнителен code, не бихте ли казали? Всеки от декларираните класове изпълнява една функция, но ние използваме куп допълнителен поддържащ code, за да я дефинираме. И така са мислor разработчиците на Java. Съответно, те въведоха набор от „функционални интерфейси“ ( @FunctionalInterface) и решиха, че сега самата Java ще „мисли“, оставяйки само важните неща, за които да се тревожим:

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

A Supplierконсумативи. Няма параметри, но връща нещо. Ето How доставя нещата. A Consumerконсумира. Той приема нещо като вход (аргумент) и прави нещо с него. Аргументът е това, което консумира. Тогава също имаме Function. Той приема входове (аргументи), прави нещо и връща нещо. Виждате, че ние активно използваме генерични лекарства. Ако не сте сигурни, можете да получите опреснителна информация, като прочетете „ Generics в Java: How да използвате ъглови скоби на практика “.

CompletableFuture

CompletableFutureМина време и в Java 1.8 се появи нов клас, наречен . Той реализира Futureинтерфейса, т.е. задачите ни ще бъдат изпълнени в бъдеще и можем да се обаждаме, get()за да получим резултата. Но също така реализира CompletionStageинтерфейса. Името казва всичко: това е определен етап от набор от изчисления. Кратко въведение в темата можете да намерите в прегледа тук: Въведение в CompletionStage и CompletableFuture. Да минем направо по същество. Нека да разгледаме списъка с налични статични методи, които ще ни помогнат да започнем: Ето опции за използването им:

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

Ако изпълним този code, ще видим, че създаването на CompletableFutureсъщо включва стартиране на цял конвейер. Следователно, с известна прorка със SteamAPI от Java8, тук намираме разликата между тези подходи. Например:

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

Това е пример за Stream API на Java 8. Ако стартирате този code, ще видите, че „Изпълнено“ няма да се покаже. С други думи, когато се създаде поток в Java, потокът не започва веднага. Вместо това, той чака някой да иска стойност от него. Но CompletableFutureзапочва да изпълнява тръбопровода веднага, без да чака някой да го попита за стойност. Мисля, че това е важно да се разбере. И така, имаме CompletableFuture. Как можем да направим тръбопровод (or верига) и Howви механизми имаме? Спомнете си онези функционални интерфейси, за които писахме по-рано.

• Имаме Function, което взема A и връща B. Има един метод: apply().
• Имаме Consumer, което взема A и не връща нищо (Void). Има един единствен метод: accept().
• Имаме Runnable, който работи в нишката и не взема нищо и не връща нищо. Има един единствен метод: run().

Следващото нещо, което трябва да запомните е, че CompletableFutureизползва Runnable, Consumersи Functionsв работата си. Съответно, винаги можете да знаете, че можете да направите следното с 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);
}

Методите thenRun(), thenApply(), и thenAccept()имат "Async" версии. Това означава, че тези етапи ще бъдат завършени на различна нишка. Тази нишка ще бъде взета от специален пул — така че няма да знаем предварително дали ще бъде нова or стара нишка. Всичко зависи от това колко изчислително интензивни са задачите. В допълнение към тези методи има още три интересни възможности. За по-голяма яснота нека си представим, че имаме определена услуга, която получава няHowво съобщение отнякъде — и това отнема време:

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

Сега, нека да разгледаме другите способности, които CompletableFutureпредоставя. Можем да комбинираме резултата от a CompletableFutureс резултата от друг 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();

Имайте предвид, че нишките са демонни нишки по подразбиране, така че за яснота използваме get()да изчакаме резултата. Не само можем да комбинираме CompletableFutures, но и да върнем CompletableFuture:

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

Тук искам да отбележа, че CompletableFuture.completedFuture()методът е използван за краткост. Този метод не създава нова нишка, така че останалата част от конвейера ще бъде изпълнена на същата нишка, където completedFutureе била извикана. Има и thenAcceptBoth()метод. Той е много подобен на accept(), но ако thenAccept()приеме a Consumer, thenAcceptBoth()приема друг CompletableStage+ BiConsumerкато вход, т.е. a consumerкойто приема 2 източника instead of един. Има още една интересна възможност, предлагана от методи, чието име включва думата „Или“: Тези методи приемат алтернатива CompletableStageи се изпълняват на този CompletableStage, който трябва да бъде изпълнен първи. И накрая, искам да завърша този преглед с друга интересна функция на CompletableFuture: обработка на грешки.

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

Този code няма да направи нищо, защото ще има изключение и нищо друго няма да се случи. Но като разкоментираме изявлението „по изключение“, ние дефинираме очакваното поведение. Говорейки за CompletableFuture, препоръчвам ви да гледате и следното видео:

• Въведение в CompletableFuture

По мое скромно мнение това са сред най-обясняващите видеоклипове в интернет. Те трябва да изяснят How работи всичко това, Howъв набор от инструменти имаме на разположение и защо е необходимо всичко това.

Заключение

Надяваме се, че вече е ясно How можете да използвате нишки, за да получите изчисления, след като са завършени. Допълнителен материал:

• Java Concurrency (Многопоточност) — CompletableFuture Explained
• Въведение в CompletionStage и CompletableFuture
• Паралелност и подобрения в Java 8: Част 2
• Ръководство за CompletableFuture

По-добре заедно: Java и клас Thread. Част I — Нишки за изпълнение По-добре заедно: Java и класът Thread. Част II — По-добра синхронизация заедно: Java и класът Thread. Част III — Взаимодействието е по-добро заедно: Java и класът Thread. Част V — Executor, ThreadPool, Fork/Join Better заедно: Java и класът Thread. Част VI — Изстрелвай!