Многопоточное программирование в Java

Вы можете использовать метод thenApply для обработки и преобразования результата CompletableFuture, когда он поступит.

Этот метод принимает реализацию интерфейса Function как аргумент.

Интерфейс Function содержит метод, который принимает аргумент и возвращает результат.

Если вы не хотите возвращать что-либо из своей функции обратного вызова и просто хотите запустить некий код после завершения Future, вы можете использовать методы thenAccept и thenRun. Эти методы используются в качестве последнего обратного вызова в цепочке обратных вызовов.

Метод thenAccept принимает реализацию интерфейса Consumer, который имеет метод accept, принимающий аргумент и ничего не возвращающий.

Этот метод имеет доступ к результату CompletableFuture.

Метод thenRun принимает объект Runnable и не имеет доступа к результату Future.

У всех этих метод есть вариация Async, которая запускает код в отдельном потоке.

Вы также можете создать цепочку вложенных CompletableFuture.

Здесь вы сначала получаете идентификатор клиента, а затем передаете его методу thenCompose и возвращаете баланс клиента.

Теперь, если вы хотите вычислить две задачи независимо друг от друга, а затем что-то вычислить с использованием их результатов, вместо метода thenCompose нужно применить метод thenCombine.

В этом примере мы независимо вычисляет ширину и высоту, а затем вычисляем площадь.

Мы использовали методы thenCompose и thenCombine чтобы соединить и объединить два CompletableFutures вместе.

Теперь, если вы хотите совместить произвольное число CompletableFutures, для этого можно использовать методы allOf и anyOf.

Далее, если у нас появилась ошибка в цепочке вызовов, мы можем легко ее обработать, и ошибка не будет распространяться дальше в цепочке обратных вызовов.

ManagedBlocker

Фреймворк Fork/Join дает возможность использовать один общий пул потоков ForkJoinPool, который предварительно сконфигурирован JVM и используется для распараллеливания потоков и выполнения задач, заданных, например, с помощью CompletableFuture.supplyAsync и так далее.

Это звучит хорошо.

Однако то, что пул потоков ForkJoinPool является общим, означает совместное использование всеми компонентами, работающими в одном JVM-процессе.

Если вы загрязните пул потоков блокирующими или длительными задачами, вы можете остановить работу всего процесса JVM, который у вас есть.

Однако ForkJoinPool был разработан с идеей о том, что некоторые задачи могут блокировать рабочие потоки, поэтому он содержит API для обработки таких случаев блокировки.

Для этого используется интерфейс ManagedBlocker, обеспечивающий способ сообщить ForkJoinPool, что он должен расширить свой параллелизм, чтобы компенсировать потенциальные заблокированные рабочие потоки.

Интерфейс ManagedBlocker предоставляет два метода.

Метод isReleasable должен возвращать true, если блокировка не требуется.

Метод block блокирует текущий поток, если это необходимо.

Эти действия выполняются любым потоком, вызывающим метод ForkJoinPool.managedBlock.

Этот метод запускает блокирующую задачу.

При этом активируется запасной поток, если это необходимо, чтобы обеспечить достаточный параллелизм, когда текущий поток заблокирован.

Этот метод также неоднократно вызывает методы интерфейса ManagedBlocker isReleasable и block, пока оба метода не вернут true.

Каждому вызову метода block предшествует вызов метода isReleasable, который возвращает false.

При блокировке, будет создан новый поток внутри пула, чтобы поддерживать параллелизм пока работает блокировка.

Как только блок с управляемым блокированием завершается, уровень параллелизма становится слишком высоким, поэтому, когда другой поток завершает выполнение своей задачи, вместо того чтобы ждать следующую задачу из очереди задач, он отбрасывается, тем самым возвращая параллелизм обратно к своему первоначальному уровню.

В этом примере мы блокируем рабочий поток вызовом метода Thread.sleep.

Для динамического расширения пула потоков мы можем обернуть этот вызов в ManagedBlocker.

Для этого мы создадим класс Sleeper, расширяющий интерфейс ManagedBlocker.

Здесь при вызове метода managedBlock, будет вызван метод isReleasable, он вернет false.

Поэтому будет вызван метод block и текущий поток заснет.

При этом в пуле будет создан новый поток.

Как только текущий поток проснется, метод isReleasable начнет возвращать true, это просигнализирует пулу потоков, что один поток можно убрать.

Для такой блокировки, мы в нашей задаче вызываем метод managedBlock, в который передаем экземпляр созданного класса с нужными параметрами.

Потокобезопасные коллекции

Фреймворк Collection, с его интерфейсами List, Set, и Map, обеспечивает потоковую безопасность с помощью синхронизированных оболочек.

Где используются синхронизированные блоки кода.

Однако эти потокобезопасные коллекции все еще имеют некоторые недостатки.

Например, необходимо блокировать коллекцию при ее итерации, иначе вы рискуете получить исключение ConcurrentModificationException.