Reactive Streams

Reactive Streams — это инициатива по разработке стандарта для асинхронной обработки потоков данных с неблокирующим противодавлением (backpressure). Основная цель Reactive Streams — определение минимального набора интерфейсов, методов и протоколов, необходимых для достижения совместимости между различными библиотеками и фреймворками, реализующими реактивное программирование. Стандарт был создан для решения проблемы эффективной передачи данных между компонентами системы в условиях, когда источник данных (Publisher) может производить элементы быстрее, чем потребитель (Subscriber) способен их обрабатывать, без использования блокировок и переполнения буферов.

История

Инициатива Reactive Streams была официально запущена в 2013 году. Её создателями выступили инженеры из компаний Typesafe (ныне Lightbend), Netflix, Pivotal и Red Hat. Основным стимулом для разработки стала необходимость преодолеть ограничения традиционных моделей обработки потоков данных, таких как Java Stream API, которые не поддерживали асинхронность и противодавление на уровне спецификации.

Первая версия спецификации (1.0.0) была опубликована в апреле 2015 года. В 2017 году Reactive Streams были включены в стандартную библиотеку Java 9 в виде модуля java.util.concurrent.Flow, который реализует интерфейсы, совместимые со спецификацией Reactive Streams. Это позволило различным библиотекам (например, RxJava, Project Reactor, Akka Streams) взаимодействовать друг с другом на едином протоколе.

Основные принципы

Спецификация Reactive Streams базируется на четырёх ключевых принципах, зафиксированных в «Манифесте реактивных систем» (Reactive Manifesto), но сфокусирована именно на асинхронной передаче данных:

1. Асинхронность: Все операции между компонентами выполняются без блокировки потоков. Передача сигналов (onNext, onError, onComplete) происходит асинхронно, что позволяет эффективно использовать ресурсы системы.
2. Неблокирующее противодавление (Non-blocking Backpressure): Это центральная концепция Reactive Streams. Подписчик (Subscriber) может сообщить издателю (Publisher) о своей готовности принять определённое количество элементов. Издатель не может отправлять данные быстрее, чем подписчик может их обработать. Механизм противодавления реализуется через метод request(long n), который подписчик вызывает для запроса следующей порции данных.
3. Обработка ошибок: Спецификация определяет строгие правила обработки ошибок. Любая ошибка, возникшая в процессе передачи данных, должна быть доставлена подписчику через сигнал onError. После этого подписка считается завершённой, и дальнейшая передача данных прекращается. Исключения, возникающие в методах подписчика, также обрабатываются специальным образом.
4. Отмена подписки: Подписчик может в любой момент отменить подписку, вызвав метод cancel() у объекта Subscription. Это приводит к немедленному прекращению получения данных издателем.

Архитектура и компоненты

Спецификация Reactive Streams определяет четыре основных интерфейса, которые должны быть реализованы для обеспечения совместимости:

Publisher (Издатель)

Publisher — это источник данных. Он производит элементы (объекты) и отправляет их подписчикам. Основной метод — subscribe(Subscriber<? super T> subscriber). При вызове этого метода Publisher обязан инициировать процесс подписки, который включает вызов метода onSubscribe(Subscription s) у подписчика.

Subscriber (Подписчик)

Subscriber — это потребитель данных. Он получает элементы от Publisher и обрабатывает их. Subscriber содержит четыре метода, которые Publisher может вызывать в строго определённом порядке:

• onSubscribe(Subscription s) — вызывается один раз при подписке. Передаёт объект Subscription, который позволяет Subscriber управлять потоком данных (запрашивать элементы или отменять подписку).
• onNext(T item) — вызывается для каждого нового элемента данных. Subscriber обрабатывает элемент.
• onError(Throwable t) — вызывается при возникновении неисправимой ошибки. После этого сигнала подписка завершается.
• onComplete() — вызывается, когда Publisher завершил отправку всех элементов. После этого сигнала подписка завершается.

Subscription (Подписка)

Subscription — это объект, представляющий собой связь между Publisher и Subscriber. Он предоставляет два метода:

• request(long n) — Subscriber вызывает этот метод, чтобы запросить у Publisher следующую порцию данных. Параметр n указывает количество запрашиваемых элементов. Это основной механизм управления противодавлением.
• cancel() — Subscriber вызывает этот метод, чтобы отменить подписку и прекратить получение данных.

Processor (Обработчик)

Processor — это компонент, который одновременно является и Publisher, и Subscriber. Он получает данные от одного Publisher, преобразует их и передаёт другому Subscriber. Processor используется для создания цепочек обработки данных (pipeline). Спецификация Reactive Streams не накладывает строгих ограничений на реализацию Processor, но требует, чтобы он корректно обрабатывал противодавление и ошибки.

Правила спецификации

Спецификация Reactive Streams содержит набор строгих правил, которые должны соблюдать все реализации. Ключевые правила включают:

• Порядок вызова: Сигналы onNext, onError и onComplete не могут вызываться одновременно. Они должны следовать в строгом порядке: onSubscribe -> (0 или более onNext) -> (onError или onComplete).
• Ограничение на количество запросов: Subscriber не может запрашивать больше элементов, чем он может обработать. Publisher не может отправлять больше элементов, чем было запрошено Subscriber.
• Обработка исключений: Если метод onNext или onSubscribe подписчика выбрасывает исключение, Publisher должен вызвать onError и завершить подписку. Если исключение выбрасывается в onError или onComplete, поведение не определено, и подписка считается завершённой.
• Потокобезопасность: Все методы должны быть потокобезопасными, но спецификация не требует, чтобы Publisher вызывал методы Subscriber из одного и того же потока.

Реализации и примеры использования

Спецификация Reactive Streams не является библиотекой или фреймворком. Она определяет только интерфейсы и правила. На практике Reactive Streams используются в нескольких популярных библиотеках и фреймворках:

• RxJava (Reactive Extensions for Java) — одна из первых реализаций реактивного программирования, поддерживающая Reactive Streams.
• Project Reactor — реактивная библиотека, лежащая в основе Spring WebFlux. Она полностью совместима со спецификацией Reactive Streams.
• Akka Streams — реализация для платформы Akka, использующая модель акторов.
• Java 9 Flow API (java.util.concurrent.Flow) — встроенная реализация интерфейсов Reactive Streams в стандартной библиотеке Java.
• Vert.x — набор инструментов для создания реактивных приложений, поддерживает Reactive Streams через свой модуль vertx-rx.

Пример использования в Java 9 Flow API:

```java import java.util.concurrent.Flow.*;

public class SimpleExample { public static void main(String[] args) throws InterruptedException { // Создаем Publisher, который генерирует числа от 1 до 5 SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

// Создаем Subscriber Subscriber<Integer> subscriber = new Subscriber<>() { private Subscription subscription;

@Override public void onSubscribe(Subscription subscription) { this.subscription = subscription; // Запрашиваем первый элемент subscription.request(1); }

@Override public void onNext(Integer item) { System.out.println("Получено: " + item); // Запрашиваем следующий элемент subscription.request(1); }

@Override public void onError(Throwable throwable) { System.err.println("Ошибка: " + throwable.getMessage()); }

@Override public void onComplete() { System.out.println("Завершено"); } };

// Подписываем Subscriber на Publisher publisher.subscribe(subscriber);

// Публикуем данные publisher.submit(1); publisher.submit(2); publisher.submit(3); publisher.submit(4); publisher.submit(5);

// Закрываем Publisher publisher.close();

// Ждем завершения Thread.sleep(100); } } ```

Критика и ограничения

Несмотря на широкое распространение, спецификация Reactive Streams подвергается критике по нескольким причинам:

• Сложность реализации: Соблюдение всех правил спецификации (особенно в части потокобезопасности и обработки ошибок) требует значительных усилий от разработчиков библиотек.
• Отсутствие поддержки мультиплексирования: Спецификация не определяет, как обрабатывать случаи, когда один Publisher обслуживает множество Subscribers с разными запросами на количество элементов. Это оставлено на усмотрение реализаций.
• Ограниченная поддержка в некоторых языках: Reactive Streams изначально разрабатывались для Java и JVM-языков. Хотя существуют реализации для других языков (например, JavaScript, .NET), они не всегда полностью соответствуют спецификации.
• Необходимость обучения: Разработчики, привыкшие к синхронным моделям, часто испытывают трудности с пониманием концепций противодавления и асинхронной обработки потоков.

Источники

• Reactive Streams Specification (Version 1.0.3) — официальная спецификация.
• Reactive Manifesto — документ, описывающий принципы реактивных систем.
• Java Platform, Standard Edition (Java SE) 9 — документация по модулю java.util.concurrent.Flow.
• Документация Project Reactor, RxJava, Akka Streams.