Как (не) использовать реактивные потоки в Java 9+

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

Результаты тестирования издателя TCK на SimplePublisher.

На самом деле, прошел только один тестовый случай; у всех остальных проблемы. Это ясно демонстрирует, что тривиальная реализация не является правильной.

Числа в именах тестовых примеров относятся к соответствующим элементам в спецификации Reactive Streams,  где вы можете дополнительно изучить концепции, лежащие в основе этих требований.

Оказывается, что большинство проблем можно устранить с помощью пары небольших изменений, а именно:

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

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

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

Поскольку реализация является синхронной, существует проблема с неограниченной рекурсией, возникающая из-за того, что подписка request()вызывает абонента onNext(), где, в свою очередь, абонент request()снова звонит и т. Д.

Другая серьезная проблема связана с обработкой бесконечного спроса (т. Е. Подписчик запрашивает Long.MAX_VALUEэлементы, возможно, пару раз). Если вы здесь недостаточно осторожны, вы можете либо создать слишком много потоков, либо переполнить какое-то longзначение, чтобы вы могли сохранить накопленный спрос.

Не пробуй это дома

Суть приведенного выше примера заключается в том, что реактивные компоненты на самом деле не тривиальны для правильной реализации. Итак, если вы не создаете еще одну реализацию Reactive Streams, вам не следует реализовывать их самостоятельно, а использовать существующие реализации, которые проверены с помощью TCK.

И если вы все равно решите написать свою собственную реализацию, убедитесь, что понимаете все детали спецификации и не забудьте запустить TCK для своего кода.

Цель новых интерфейсов

Итак, для чего нужны интерфейсы, спросите себя? Фактическая цель включения их в JDK — предоставить то, что называется уровнем интерфейса поставщика услуг (или SPI). В конечном итоге это должно служить уровнем объединения для различных компонентов, которые имеют реактивную и потоковую природу, но могут предоставлять свои собственные пользовательские API и, следовательно, не смогут взаимодействовать с другими аналогичными реализациями.

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

Существующие потоковые абстракции

Итак, какие потоковые абстракции уже есть в JDK (что означает потоковую обработку больших, возможно бесконечных, объемов данных, порций за порциями, без предварительного считывания всех данных в память)? К ним относятся:

• java.io.InputStream/OutputStream
• java.util.Iterator
• java.nio.channels.*
• javax.servlet.ReadListener/WriteListener
• java.sql.ResultSet
• java.util.Stream
• java.util.concurrent.Flow.*

Хотя все вышеперечисленные абстракции демонстрируют какое-то потоковое поведение, им не хватает общего API, который бы позволял вам легко их подключать, например, использовать a Publisherдля чтения данных из одного файла и a Subscriberдля записи в другой.

Преимущество такого объединяющего слоя заключается в возможности использовать один вызов:

publisher.subscribe(subscriber)

Используйте это для обработки всех скрытых сложностей обработки реактивного потока (таких как противодавление и сигнализация).

На пути к идеальному миру

Каковы могут быть возможные результаты использования различных абстракций с общими интерфейсами? Давайте посмотрим на несколько примеров.

Минимальный набор операций

Текущая поддержка Reactive Streams в JDK ограничена четырьмя интерфейсами, описанными ранее. Если вы когда-либо использовали какую-либо реактивную библиотеку —   Akka Streams , RxJava или Project Reactor   — вы знаете, что их сила заключается в различных потоковых комбинаторах (например, mapили, filterесли назвать самые простые), доступных из коробки. Эти комбинаторы, однако, отсутствуют в JDK, хотя вы, вероятно, ожидаете, что по крайней мере пара из них будет доступна.

Чтобы решить эту проблему, Lightbend предложил POC Reactive Streams Utilities   - библиотеку со встроенными базовыми операциями и возможностью предоставления более сложных из них в качестве плагинов, делегирующих существующей реализации, определяемой системным параметром JVM. как:

-Djava.flow.provider=akka

HTTP

Как мы получаем файл, загруженный через HTTP, и загружаем его куда-то еще, конечно же, в реактивном режиме?

Начиная с версии 3.1 сервлета существует асинхронный ввод-вывод сервлета. Кроме того, начиная с JDK 9, появился новый HTTP-клиент (который был в jdk.incubating.httpмодуле в Java 9/10, но считается стабильным с Java 11 и далее). Помимо более приятного API, новый клиент также поддерживает Reactive Streams в качестве ввода / вывода. Среди прочего, он предоставляет POST(Publisher<ByteBuffer>)метод.

Теперь, если HttpServletRequestбы издатель предоставил выставить тело запроса, загрузка полученного файла стала бы:

POST(BodyPublisher.fromPublisher(req.getPublisher())

Это происходит со всеми реактивными функциями под капотом — просто с помощью этой единственной строки кода.

Доступ к базе данных

Когда речь заходит об универсальном способе доступа к реляционной базе данных реактивным способом, появилась некоторая надежда, связанная с API асинхронного доступа к базе данных (ADBA) , который, к сожалению, до сих пор не дошел до JDK.

Существует также R2DBC   —   попытка внедрить API реактивного программирования в реляционные хранилища данных . В настоящее время он поддерживает H2 и Postgres и прекрасно работает с Spring Data JPA , что может быть преимуществом, которое помогает при более широком внедрении.

Затем существуют некоторые специфичные для производителя асинхронные драйверы. Но мы все еще не нашли идеальное решение, которое позволило бы вам сделать что-то вроде:

Publisher<User> users = entityManager    
.createQuery("select u from users")    
  .getResultPublisher()

public class IntegrationApp {

    public static void main(String[] args) {
        var reactorPublisher = reactorPublisher();
        var akkaStreamsProcessor = akkaStreamsProcessor();

        reactorPublisher.subscribe(akkaStreamsProcessor);

        Flowable
            .fromPublisher(FlowAdapters.toProcessor(akkaStreamsProcessor))
            .subscribe(System.out::println);
    }

    private static Publisher<Long> reactorPublisher() {
        var numberFlux = Flux.interval(Duration.ofSeconds(1));
        return JdkFlowAdapter.publisherToFlowPublisher(numberFlux);
    }

    private static Processor<Long, Long> akkaStreamsProcessor() {
        var negatingFlow = Flow.of(Long.class).map(i -> -i);
        return JavaFlowSupport.Flow.toProcessor(negatingFlow).run(materializer);
    }

    private static ActorSystem actorSystem = ActorSystem.create();
    private static ActorMaterializer materializer = ActorMaterializer.create(actorSystem);
}