Реактивное программирование Java

Реактивный поток описывает асинхронное поступление от нуля до многих элементов и завершение одним из двух сигналов: complete или error. Главная особенность Reactive Streams — получатель явно сообщает, сколько элементов готов обработать.

Контракт Reactive Streams

В протоколе участвуют четыре роли:

• Publisher<T> публикует элементы;
• Subscriber<T> принимает сигналы;
• Subscription связывает их и принимает request(n) или cancel();
• Processor<T, R> одновременно subscriber и publisher.

Порядок сигналов для одного subscriber:

onSubscribe(subscription)
onNext(item) × 0..N
onComplete()

или:

onSubscribe(subscription)
onNext(item) × 0..N
onError(error)

После terminal-сигнала новые элементы запрещены. onError — часть протокола, а не исключение, которое можно поймать вокруг асинхронного subscribe.

Backpressure

Subscriber вызывает request(n), тем самым выдавая publisher разрешение отправить не более n элементов. Это защищает медленного потребителя от неограниченного push-потока, если источник умеет соблюдать спрос.

final class BatchSubscriber<T> implements Flow.Subscriber<T> {
    private Flow.Subscription subscription;
    private int received;
 
    @Override
    public void onSubscribe(Flow.Subscription value) {
        subscription = value;
        subscription.request(20);
    }
 
    @Override
    public void onNext(T item) {
        handle(item);
        if (++received == 20) {
            received = 0;
            subscription.request(20);
        }
    }
 
    @Override public void onError(Throwable error) { report(error); }
    @Override public void onComplete() { finish(); }
}

java.util.concurrent.Flow содержит интерфейсы, совместимые по идее с Reactive Streams, но не даёт богатого набора операторов.

Cold и hot publisher

Cold publisher заново производит последовательность для каждого subscriber. Запрос к файлу или БД часто cold: две подписки могут выполнить работу дважды.

Hot publisher существует независимо от конкретного подписчика. Поздний подписчик может не увидеть ранние события. Для него нужно заранее решить, что делать с медленными и отсутствующими потребителями: буферизовать, отбрасывать, хранить последнее значение или отключать.

Асинхронность не равна параллельности

Reactive API не обещает новый поток для каждого оператора. Пока scheduler не переключён явно, цепочка часто выполняется на потоке источника. Параллельность и порядок задаются отдельно.

Неблокирующий код не удерживает поток в ожидании I/O. Но если внутри цепочки вызвать блокирующий JDBC, файловую операцию или .get(), поток всё равно будет заблокирован.

Когда подход полезен

• много одновременных I/O-операций;
• источники естественно передают события;
• нужно управлять спросом и отменой;
• весь путь запроса поддерживает неблокирующие драйверы.

Для короткой CRUD-операции с JDBC reactive-обёртка не устраняет блокировку. Иногда обычный последовательный код проще и надёжнее.

Частые ошибки

• вызов subscribe() внутри бизнес-метода вместо возврата publisher;
• блокировка event loop через block(), JDBC или долгий CPU-код;
• повторная подписка к cold publisher и повтор побочного эффекта;
• бесконечный буфер вместо реального backpressure;
• потеря контекста и ошибки при переключении scheduler;
• отсутствие timeout и отмены внешнего запроса.

На собеседовании

Чем reactive stream отличается от Java Stream? Java Stream — синхронная pull-обработка конечного или вычисляемого источника; reactive stream — сигнализируемая во времени последовательность с завершением, ошибкой и управлением спросом.

Backpressure гарантирует отсутствие переполнения памяти? Нет. Оператор или внешний источник может буферизовать данные; стратегию очереди и пределы всё равно задаёт приложение.

Связанные заметки: Spring Reactor, Многопоточность в Java, Объект Future.