Java вычисления в потоке
public class MyFirstThread extends Thread < @Override public void run() < switch (getName()) < case "Thread-0" ->System.out.println("Поток-0 достает яйца из холодильника"); case "Thread-1" -> System.out.println("Поток-1 включает плиту"); case "Thread-2" -> System.out.println("Поток-2 достает сковородку и ставит на плиту."); case "Thread-3" -> System.out.println("Поток-3 зажигает огонь на плите"); case "Thread-4" -> System.out.println("Поток-4 выливает на сковороду масла"); case "Thread-5" -> System.out.println("Поток-5 разбивает яйца и выливает их на сковороду"); case "Thread-6" -> System.out.println("Поток-6 выбрасывает скорлупу в мусорное ведро"); case "Thread-7" -> System.out.println("Поток-7 снимает готовую яичницу с огня"); case "Thread-8" -> System.out.println("Поток-8 выкладывает яичницу в тарелку"); case "Thread-9" -> System.out.println("Поток-9 моет посуду"); > //System.out.println("I'm Thread! My name is " + getName()); > public static void main(String[] args) < for (int i = 0; i < 10; i++) < MyFirstThread thread = new MyFirstThread(); thread.start(); >> > «В некоторых штатах Индии вам не продадут землю сельскохозяйственного назначения, если вы не зарегистрированы как фермер. При этом вас не зарегистрируют в качестве фермера, если вы не владеете сельскохозяйственными землями». Перефразирую на реальные факты : «В IT компанию вас не возьмут на работу, если у вас нет опыта. При этом вы не получите опыт, если не работаете в IT».
Thread’ом Java не испортишь: Часть IV — Callable, Future и друзья
Мы уже рассматривали в первой части, как создаются потоки. Ещё раз вспомним. Поток — это Thread , в нём что-то запускается run , поэтому воспользуемся tutorialspoint java online compiler’ом и выполним следующий код:
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 , который возвращает результат. Кроме того, по умолчанию он throws Exception. То есть избавляет нас от необходимости на проверяемые исключения писать try-catch блоки. Уже неплохо, правда? Теперь у нас есть вместо Runnable новый task:
Но что с ним делать? Зачем нам вообще задача, выполняемая в потоке, которая возвращает результат? Очевидно, что в дальнейшем мы рассчитываем получить результат действий, которыев в будущем будут выполнены. Будущее по-английский — Future. И интерфейс есть с точно таким же именем: java.util.concurrent.Future
java.util.concurrent.Future
Интерфейс java.util.concurrent.Future описывает API для работы с задачами, результат которых мы планируем получить в будущем: методы получения результата, методы проверки статуса. Для Future нас интересует его реализация java.util.concurrent.FutureTask. То есть это Task , который будет выполнен во 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 future = new FutureTask<>(task); new Thread(future).start(); System.out.println(future.get()); > > Как видно из примера, мы получаем при помощи метода get результат из задачи task . (!)Важно , что в момент получения результата при помощи метода get выполнение становится синхронным. Как вы думаете, какой механизм тут будет использован? Правильно, нет блока синхронизации — поэтому WAITING в JVisualVM мы увидим не как monitor или wait , а как тот самый park (т.к. используется механизм LockSupport ).
Функциональные интерфейсы
Дальше пойдёт речь про классы из Java 1.8, поэтому не лишним будет сделать краткое введение. Посмотрим на следующий код:
Supplier supplier = new Supplier() < @Override public String get() < return "String"; >>; Consumer consumer = new Consumer() < @Override public void accept(String s) < System.out.println(s); >>; Function converter = new Function() < @Override public Integer apply(String s) < return Integer.valueOf(s); >>; Как же много лишнего кода, не правда ли? Каждый из объявляемых классов выполняет какую-то одну функцию, но для её описания мы используем кучу лишнего вспомогательного кода. И разработчики Java так же подумали. Поэтому, они ввели набор «функциональных интерфейсов» ( @FunctionalInterface ) и решили, что теперь Java сама будет «додумывать» за нас всё, кроме важного:
Supplier supplier = () -> "String"; Consumer consumer = s -> System.out.println(s); Function converter = s -> Integer.valueOf(s); Supplier — поставщик. Он не имеет параметров, но возвращает что-то, то есть поставляет это. Consumer — потребитель. Он принимает на вход что-то (параметр s) и с этим что-то что-то делает, то есть потребляет что-то. Есть ещё функция. Она принимает на вход что-то (параметр s ), что-то делает и возвращает что-то. Как мы видим, активно используются дженерики. В случае неуверенности можно вспомнить про них и прочитать «Теория дженериков в Java или как на практике ставить скобки».
CompletableFuture
Шло время, и в Java 1.8 появился новый класс, который зовётся CompletableFuture . Он реализует интерфейс Future , то есть наши task будут выполнены в будущем, и мы сможем выполнить get и получить результат. Но ещё он реализует некоторый CompletionStage . Из перевода уже понятно его назначение: это некий этап (Stage) каких-то вычислений. С кратким введением в тему можно ознакомиться в обзоре «Introduction to CompletionStage and CompletableFuture». Давайте перейдём сразу к делу. Посмотрим на список доступных статических методов, которые нам помогут начать: Вот варианты их использования:
import java.util.concurrent.CompletableFuture; public class App < public static void main(String []args) throws Exception < // CompletableFuture уже содержащий результат CompletableFuturecompleted; completed = CompletableFuture.completedFuture("Просто значение"); // CompletableFuture, запускающий (run) новый поток с Runnable, поэтому он Void CompletableFuture voidCompletableFuture; voidCompletableFuture = CompletableFuture.runAsync(() -> < System.out.println("run " + Thread.currentThread().getName()); >); // CompletableFuture, запускающий новый поток, результат которого возьмём у Supplier CompletableFuture supplier; supplier = CompletableFuture.supplyAsync(() -> < System.out.println("supply " + Thread.currentThread().getName()); return "Значение"; >); > > Если мы выполним этот код, то увидим, что создание CompletableFuture подразумевает запуск и всей цепочки. Поэтому при некоторой схожести со SteamAPI из Java8 в этом отличие этих подходов. Например:
List array = Arrays.asList("one", "two"); Stream stringStream = array.stream().map(value -> < System.out.println("Executed"); return value.toUpperCase(); >); - У нас есть функция ( Function ), которая принимает А и возвращает Б. Имеет единственный метод — apply (применить).
- У нас есть потребитель ( Consumer ), которая принимает А и ничего не возвращает (Void). Имеет единственный метод — accept (принять).
- У нас есть запускаемый в потоке код Runnable , который не принимает и не возвращает. Имеет единственный метод — run (запустить).
public static void main(String []args) throws Exception < AtomicLong longValue = new AtomicLong(0); Runnable task = () ->longValue.set(new Date().getTime()); Function dateConverter = (longvalue) -> new Date(longvalue); Consumer 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 . Это значит, что эти стадии будут выполнены в новом потоке. Он будет взят из особого пула, поэтому заранее неизвестно, какой поток будет, новый или прежний. Всё зависит от того, на сколько тяжёлые задачи. Помимо этих методов есть ещё три интересные возможности. Для наглядности представим, что у нас есть некий сервис, который получает какое-то сообщение откуда-то и на это требуется время:
public static class NewsService < public static String getMessage() < try < Thread.currentThread().sleep(3000); return "Message"; >catch (InterruptedException e) < throw new IllegalStateException(e); >> > Теперь, давайте посмотрим на другие возможности, которые предоставляет CompletableFuture . Мы можем объединять результат CompletableFuture с результатом другого CompletableFuture :
Supplier newsSupplier = () -> NewsService.getMessage(); CompletableFuture reader = CompletableFuture.supplyAsync(newsSupplier); CompletableFuture.completedFuture("!!") .thenCombine(reader, (a, b) -> b + a) .thenAccept(result -> System.out.println(result)) .get(); Тут стоить обратить внимание, что по умолчанию потоки будут демон-потоками, поэтому для наглядности мы используем get , чтобы дождаться результат. А ещё мы можем не только объединить (combine), но и возвращать CompletableFuture :
CompletableFuture.completedFuture(2L) .thenCompose((val) -> CompletableFuture.completedFuture(val + 2)) .thenAccept(result -> System.out.println(result)); 
Тут хочется отметить, что для краткости использован метод CompletableFuture.completedFuture . Данный метод не создаёт новый поток, поэтому остальная цепочка будет выполнена в том же потоке, в котором был вызван completedFuture . Также есть метод thenAcceptBoth . Он очень похож на accept , но если thenAccept принимает consumer , то thenAcceptBoth принимает на вход ещё один CompletableStage + BiConsumer , то есть consumer , который на вход принимает 2 источника, а не один. Есть ещё интересная возможность со словом Either : Данные методы принимают альтернативный CompletableStage и будут выполнены на том CompletableStage , который первее выполнится. И закончить этот обзор хочется ещё одной интересной возможностью CompletableFuture — обработкой ошибок.
CompletableFuture.completedFuture(2L) .thenApply((a) -> < throw new IllegalStateException("error"); >).thenApply((a) -> 3L) //.exceptionally(ex -> 0L) .thenAccept(val -> System.out.println(val));