Java for Beginner
743 subscribers
708 photos
196 videos
12 files
1.14K links
Канал от новичков для новичков!
Изучайте Java вместе с нами!
Здесь мы обмениваемся опытом и постоянно изучаем что-то новое!

Наш YouTube канал - https://www.youtube.com/@Java_Beginner-Dev

Наш канал на RUTube - https://rutube.ru/channel/37896292/
Download Telegram
Реактивное программирование

Вступление

В современном мире разработки программного обеспечения мы всё чаще сталкиваемся с задачами, где простая последовательная логика уже не справляется. Нагрузки на системы растут экспоненциально: миллионы пользователей ожидают мгновенных ответов, микросервисы обмениваются сотнями запросов в секунду, а данные льются рекой из баз данных, API и внешних сервисов.
Традиционный подход — "запрос приходит, запускается поток, ждём ответа" — начинает давать сбои: система тонет в блокировках, лишних потоках и перерасходе ресурсов.


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

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


Асинхронщина и её проблемы: от основ к пределам

Что такое асинхронность в программировании?
Это когда программа не стоит на месте, ожидая завершения одной задачи, а продолжает работать с другими. В Java это реализуется через механизмы, позволяющие запускать код параллельно. Но на пути к идеалу было много подводных камней.



Потоки: базовый, но тяжёлый инструмент

Многопоточность в Java появилась ещё в ранних версиях языка. Идея проста: вы создаёте новый поток — это как отдельный "рабочий", который выполняет код независимо от основного.


Код выглядит так:
new Thread(() -> { /* ваш код */ }).start();. 


Внутри метода run() вы пишете, что нужно сделать. На первый взгляд, это решает проблему: пока один поток ждёт ответа от сервера, другой может обрабатывать новый запрос.

Но практика показывает минусы. Каждый поток — это тяжёлый объект: он требует выделения памяти (обычно от 1 МБ на стек), системных ресурсов операционной системы и времени на создание. Если нагрузка растёт — скажем, 10 тысяч одновременных запросов, — вы не сможете создать 10 тысяч потоков: система просто исчерпает ресурсы, и всё упадёт. Ещё одна боль — контекстные переключения: когда процессор переключается между потоками, это стоит CPU-времени, иногда до микросекунд на переключение, что накапливается в большие задержки.

Чтобы смягчить это, в Java ввели ExecutorService — сервис-исполнитель, который управляет пулом потоков. Вы создаёте фиксированный пул, например, Executors.newFixedThreadPool(10), и подаёте задачи:
executor.execute(() -> { /* код */ });


Теперь потоки переиспользуются: закончил задачу — берёт следующую. Это экономит ресурсы, но не решает корень проблемы. Если в задаче есть блокирующий код — например, чтение из файла или ожидание сети, — поток всё равно "зависает" в ожидании, блокируя место в пуле. Другие задачи ждут в очереди, и под высокой нагрузкой пул исчерпывается. В итоге, асинхронность есть, но она неэффективна: ресурсы тратятся на ожидание, а не на полезную работу.



#Java #middle #Reactor
👍3
Future: обещание результата, но с подвохом

В Java 5 ввели Future — это как "чек" на будущий результат.
Вы подаёте задачу в executor и получаете объект Future, который обещает: "когда-нибудь я дам тебе ответ".


Пример:
ExecutorService executor = Executors.newFixedThreadPool(10); Future<String> future = executor.submit(() -> { Thread.sleep(1000); return "Привет"; });.


Плюс в том, что вы можете продолжать работу, не дожидаясь: пока задача крутится в фоне, основной код идёт дальше. Но чтобы забрать результат, нужно вызвать future.get(). И вот здесь засада: get() блокирует текущий поток до тех пор, пока задача не завершится. Если задача задерживается — скажем, из-за сети, — ваш поток тоже висит в ожидании. Получается, асинхронность иллюзорна: да, запуск асинхронный, но использование результата синхронное и блокирующее. Это как заказать еду по доставке, но стоять у двери, не отходя, пока курьер не приедет. Выигрыш минимален, особенно в веб-приложениях, где запросы должны обрабатываться быстро.


Ещё Future неудобен в композиции: если нужно объединить результаты нескольких задач, приходится вручную ждать каждого get(), что приводит к спагетти-коду с try-catch для ошибок и таймаутами.


CompletableFuture: цепочки действий, но без избавления от ада

Java 8 принесла CompletableFuture — улучшенную версию Future, которая позволяет строить цепочки асинхронных операций без блокировок на get(). Теперь результат можно обрабатывать через "колбэки" — функции, которые вызываются автоматически по завершении.

Пример:
CompletableFuture.supplyAsync(() -> { return "Данные"; }).thenApply(data -> { return data.toUpperCase(); }).thenAccept(System.out::println);.

Здесь supplyAsync запускает задачу асинхронно, thenApply преобразует результат (например, переводит в верхний регистр), thenAccept выводит его.


Есть методы для комбинации: thenCompose для последовательных цепочек, thenCombine для параллельного объединения результатов, handle для обработки ошибок. Это шаг вперёд: код становится declarative (описательным), вы фокусируетесь на "что сделать", а не "как ждать". Нет нужды в ручном get() — всё течёт само.

Но радость недолговечна. Когда приложение усложняется — например, нужно асинхронно запросить данные из базы, потом из внешнего API, обработать ошибки и объединить, — цепочки лямбд растут в "callback-ад" (ад колбэков): вложенные функции, которые трудно читать, отлаживать и тестировать. Один уровень — ок, но пять-шесть — и код превращается в пирамиду, где сложно отследить поток выполнения.

Ещё хуже: под капотом блокировки никуда не делись. Если в цепочке есть блокирующий вызов — например, Thread.sleep() для симуляции задержки или JDBC-драйвер, который ждёт ответа от базы, блокируя поток, — весь CompletableFuture теряет преимущество. Поток из пула всё равно занят ожиданием, и под нагрузкой система снова захлёбывается. Плюс, управление ошибками в цепочках требует осторожности: одна ошибка может сломать всю последовательность, если не обработать timely.

В итоге, CompletableFuture дал выразительный синтаксис и удобство для простых сценариев, но не решил системные проблемы: ресурсы тратятся впустую на блокировки, сложность растёт, а масштабируемость под вопросом.


#Java #middle #Reactor
👍3
Callback-ад и блокировки: кульминация проблем

Callback-ад — это когда колбэки (функции обратного вызова) наслаиваются друг на друга, делая код нечитаемым. В CompletableFuture это проявляется в глубоких цепочках: thenApply внутри thenCompose, с handle для ошибок. Отладка — кошмар: где именно сломалось? Тестирование — тоже, потому что асинхронность добавляет неопределённость в порядок выполнения.

Блокировки — это когда код "зависает" в ожидании внешнего события, не давая потоку работать с другими задачами. В Java многие библиотеки (как старые IO или JDBC) блокирующие по природе: они используют системные вызовы, которые стопорят поток. Даже в асинхронных конструкциях, если внутри лямбды такая блокировка, весь пул потоков может исчерпаться. Представьте сервер с 100 потоками: 100 запросов с задержкой — и новые ждут в очереди, вызывая таймауты.

Это приводит к неэффективности: CPU простаивает, память тратится на "спящие" потоки, а под пиковой нагрузкой система не масштабируется горизонтально.


Почему нужен новый подход: реактивное программирование

Мы дошли до предела традиционных моделей. Потоки хороши для CPU-bound задач (расчёты), но тяжёлые для IO-bound (сеть, диски). Future дал обещания, но не избавил от блокировок. CompletableFuture улучшил код, но оставил callback-ад и зависимость от неблокирующих библиотек.


Здесь на сцену выходит реактивное программирование — подход, где мы думаем в терминах потоков данных и событий, а не отдельных задач. Вместо "запрос → блокировка в потоке → ответ" мы строим конвейеры: данные приходят асинхронно по мере готовности, обработка идёт реактивно, без выделения потока на каждое ожидание. Это как перейти от конвейера с паузами к непрерывному потоку. В следующих постах разберём Reactive Streams, Flux/Mono в Project Reactor и как это решает проблемы.


#Java #middle #Reactor
👍4
Реактивное программирование

Что такое потоки данных в реактивном мире?

В реактивном программировании данные — это не статичный объект, который вы запрашиваете и ждёте. Это динамичный поток: последовательность элементов (событий), которые могут приходить в любое время, в любом количестве. Поток может быть бесконечным (как лента новостей) или конечным (как результаты поиска). Главное — обработка идёт реактивно: программа "подписывается" на поток и реагирует на каждый элемент по мере его появления, без блокировок.

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

В основе лежит спецификация Reactive Streams — стандарт, который определяет, как строить такие потоки.

Он включает четыре ключевых интерфейса:
Издатель (Publisher): источник данных. Он "публикует" элементы потока. Например, это может быть база данных, генерирующая записи по запросу.
Подписчик (Subscriber): получатель данных. Он "подписывается" на издателя и реагирует на элементы: onNext (получил элемент), onError (ошибка), onComplete (поток завершён).
Подписка (Subscription): связь между издателем и подписчиком. Через неё подписчик может запросить больше данных (request(n)) или отменить подписку (cancel). Это вводит "обратное давление" — механизм, чтобы подписчик не захлёбывался данными, если не успевает их обрабатывать.
Процессор (Processor): комбинация издателя и подписчика, для промежуточной обработки (как фильтр в конвейере).


Эти интерфейсы — основа. Они обеспечивают асинхронность без блокировок: всё работает на основе событий, а не ожиданий.


События как река: метафора и практика

Представьте реку событий: вода (данные) течёт непрерывно, иногда бурно (пиковая нагрузка), иногда спокойно. Ваша программа — не плотина, которая блокирует поток, а турбина, которая генерирует энергию по мере течения. Если река слишком быстрая — турбина сигнализирует "замедлить" (обратное давление), чтобы не перегрузиться.

В Project Reactor это реализовано через два типа потоков:
Mono: поток для нуля или одного элемента. Идеален для одиночных операций, как HTTP-запрос, возвращающий один ответ.

Пример:
Mono<string> mono = Mono.just("Привет из реки");  // Создаём простой Mono с одним элементом
mono.subscribe(
value -> System.out.println("Получено: " + value), // onNext: реакция на элемент
error -> System.err.println("Ошибка: " + error), // onError
() -> System.out.println("Завершено") // onComplete
);

Здесь subscribe — это подписка. Mono "течёт" асинхронно: если элемент готов — срабатывает onNext, потом onComplete. Нет блокировок: код continue после subscribe.


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

Пример:
Flux<integer> flux = Flux.range(1, 5);   // Поток чисел от 1 до 5
flux.subscribe(
value -> System.out.println("Элемент: " + value),
error -> System.err.println("Ошибка: " + error),
() -> System.out.println("Поток завершён")
);


Flux "выдаёт" элементы по одному: 1, 2, 3... Это как река: элементы приходят последовательно, но асинхронно.

Почему это лучше традиционных подходов?
В CompletableFuture вы строите цепочки, но рискуете блокировками внутри.
В реактивном стиле всё неблокирующее: используйте операторы вроде map (преобразовать элемент), filter (отфильтровать), flatMap (развернуть в подпотоки).


Пример цепочки:
Flux.fromIterable(List.of("яблоко", "банан", "вишня"))
.filter(fruit -> fruit.startsWith("б")) // Фильтруем
.map(String::toUpperCase) // Преобразуем
.subscribe(System.out::println); // Подписываемся

Результат: "БАНАН". Всё течёт как конвейер, без callback-ада: код читаем, как последовательный, но работает асинхронно.



#Java #middle #Reactor #data_stream
👍3
Обратное давление: контроль

Одна из ключевых фишек — backpressure (обратное давление). В традиционных системах, если производитель данных быстрее потребителя, буфер переполняется, и система падает. В Reactive Streams подписчик через Subscription.request(n) говорит: "Дай мне n элементов". Издатель выдаёт ровно столько, сколько запрошено. Это как шлюзы на реке: предотвращают наводнение.


Пример в Flux:
Flux<integer> fastFlux = Flux.range(1, 1000); // Быстрый источник
fastFlux.subscribe(new BaseSubscriber<integer>() {

@Override
protected void hookOnSubscribe(Subscription subscription) {
request(10); // Запрашиваем только 10 элементов сначала
}

@Override
protected void hookOnNext(Integer value) {
System.out.println(value);
if (value % 10 == 0) request(10); // Запрашиваем ещё по мере обработки
}
});


Подписчик контролирует темп, избегая перегрузки.


Почему это новый подход, который нам нужен?

Реактивное программирование не просто добавляет инструменты — оно меняет мышление. Вместо "императивного" кода (делай то, жди это), мы пишем "декларативно": опиши, как реагировать на поток. Это масштабируется: на сервере с 4 ядрами можно обрабатывать тысячи подключений, потому что нет блокирующих потоков. Библиотеки вроде Reactor интегрируются с неблокирующими драйверами (например, reactive JDBC или WebFlux для веб), решая боли блокировок.


#Java #middle #Reactor #data_stream
👍4
Реактивное программирование

Концепции реактивного программирования: Push vs Pull — кто управляет данными

Сегодня разберём модели Push (отправку данных) и Pull (получение данных), почему одна из них идеальна для реактивного стиля, и как это решает проблемы из первого поста: от тяжёлых потоков до callback-ада.

Представьте: у вас есть конвейер на фабрике. В pull-модели рабочий сам подходит к предыдущему этапу и "берет" деталь, когда ему нужно. В push-модели предыдущий этап "отправляет" деталь дальше, как только она готова, не спрашивая.
Какая модель лучше? Зависит от сценария. В программировании это то же самое: pull подходит для предсказуемых, контролируемых потоков, а push — для динамичных, событийных, где данные приходят непредсказуемо (сеть, пользователи, сенсоры).

Реактивное программирование строится на push — и вот почему это революция.



Pull-модель: вы контролируете темп, но рискуете блокировками

В традиционном императивном программировании (когда код выполняется шаг за шагом, как рецепт) данные обычно "получаются" по требованию. Это pull-модель: потребитель (ваш код) сам запрашивает данные, когда готов их обработать.

Пример — чтение из итератора в Java:
List<String> fruits = Arrays.asList("яблоко", "банан", "вишня");

Iterator<String> iterator = fruits.iterator();

while (iterator.hasNext()) {
String fruit = iterator.next(); // "Вытягиваем" следующий элемент
System.out.println(fruit.toUpperCase());
}


Здесь вы контролируете процесс: hasNext() проверяет наличие, next() получает элемент. Это удобно для локальных, синхронных данных — вы знаете, сколько элементов, и ничего не ждёте. Но под капотом это синхронно: если данные в потоке (файл, сеть), next() может заблокироваться, как в старых потоках. Поток висит в ожидании, ресурсы тратятся впустую.

В асинхронных сценариях pull эволюционировал в Future или CompletableFuture: вы "запрашиваете" результат через get() или цепочки, но контроль остаётся у вас.
Проблема: если источник данных медленный (БД, API), ваш pull-запрос блокирует или создаёт callback-ад. Под нагрузкой — тысячи pull-запросов — система не масштабируется, потому что каждый требует ресурса (потока). Это как толпа в очереди: каждый тянет за своим, но если касса одна, все стоят.


Ещё минус: pull не справляется с "лишними" данными. Если источник генерирует 1 млн событий в секунду, а вы получаете по одному — либо перегрузка буфера, либо вы не успеваете. Нет встроенного механизма, чтобы сказать "замедлись".


#Java #middle #Reactor #Push #Flux
👍3
Push-модель: данные приходят сами, реактивно и эффективно

Теперь изменим: в push-модели источник "отправляет" данные потребителю, как только они готовы, без запросов. Потребитель пассивен — он подписывается и реагирует.
Это основа реактивного программирования: события push'атся асинхронно, без блокировок. Контроль переходит к источнику, но с обратным давлением — подписчик может сказать "хватит на время".


В Reactive Streams это реализовано через Publisher и Subscriber: издатель толкает onNext(элемент), подписчик реагирует сразу.

Пример с Flux в Project Reactor (push-стиль):
Flux<String> pushFlux = Flux.fromIterable(Arrays.asList("яблоко", "банан", "вишня"))
.map(String::toUpperCase); // Преобразование в потоке

pushFlux.subscribe(
fruit -> System.out.println("Получено из push: " + fruit), // Реакция на толкание
Throwable::printStackTrace, // Обработка ошибок
() -> System.out.println("Push завершён")
);


Здесь Flux — издатель, который отправляет элементы по мере готовности. subscribe() — подписка, и элементы приходят автоматически: "ЯБЛОКО", "БАНАН"... Нет next() — нет pull.

Если источник асинхронный, например, чтение из сети:
WebClient.create()
.get()
.uri("https://api.example.com/fruits")
.retrieve()
.bodyToFlux(String.class) // Flux толкает строки из ответа
.subscribe(fruit -> System.out.println("Push из API: " + fruit));


Данные push'атся по мере поступления от сервера — без блокировок.
Reactor использует неблокирующий IO (на базе Netty), так что поток не висит: один event-loop-цикл (цикл обработки событий) обслуживает тысячи подписок.

Почему push лучше для реактивности?
Во-первых, эффективность: нет лишних проверок hasNext().
Во-вторых, естественность для событий: клик мыши, сообщение в чате — это push по природе, они приходят сами.
В-третьих, масштабируемость: тысячи подписчиков на один издатель — ок, потому что push идёт через события, а не потоки на каждого.



Гибридные сценарии: когда mix работает

На практике модели смешиваются. В Reactor Flux может имитировать pull через операторы вроде buffer() или take(), но основа — push.

Пример: pull из локального списка, но push в сеть:
Flux.fromIterable(fruits)
.map(String::toUpperCase)
.flatMap(fruit -> sendToApi(fruit)) // flatMap толкает в асинхронный API
.subscribe(result -> System.out.println("Ответ: " + result));


Здесь локальный pull (fromIterable) переходит в push (flatMap для API).
Это гибкость: используйте pull для контроля, push для асинхронности. Но важно избегать блокировок:
Reactor проверяет и предупреждает, если в лямбде блокирующий код (onBlock()).

Ещё пример из реальной жизни: стриминг видео в Netflix. Pull — когда пользователь сам получает фреймы, но под нагрузкой лагает. Push — сервер отдает фреймы по мере готовности, с буферизацией. Реактивные библиотеки (как RxJava) позволяют строить такие конвейеры.



Почему Push — ключ к новому подходу в реактивном программировании

Возвращаясь к проблемам: потоки тяжёлые, потому что pull требует ожидания; Future блокирует на get(), потому что это pull в асинхронной обёртке; CompletableFuture даёт цепочки, но push-подход в нём слаб (колбэки — это мини-push, но без полного контроля).

Реактивный push меняет всё:
данные текут как река, вы реагируете без ожидания, ресурсы на минимуме. Системы становятся resilient (устойчивыми): если один поток сломается, другие продолжают. Под нагрузкой — горизонтальное масштабирование без боли.

#Java #middle #Reactor #Push #Flux
👍3
Реактивное программирование

Концепции реактивного программирования: Reactive Streams API — Publisher и Subscriber

Сегодня разберём Reactive Streams API — это спецификация, которая лежит в сердце реактивного программирования на Java. Она не просто набор интерфейсов, а рамка для построения асинхронных потоков данных с контролем над ними. Представьте это как правила дорожного движения для потока событий: без них — хаос и пробки, с ними — плавный поток.

Reactive Streams API решает ключевые проблемы из первого поста: вместо тяжёлых потоков и callback-ада, мы получаем унифицированный способ обмена данными между компонентами. Это не библиотека, а интерфейсы, которые реализуют фреймворки вроде Project Reactor или RxJava. Они обеспечивают совместимость: один компонент от Reactor может работать с другим от Akka Streams.


Что такое Reactive Streams API


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


Это подводит нас ближе к реактивному мышлению: вместо "запроси и жди" (как в Future.get()), мы "подпишись и реагируй". Под нагрузкой это экономит ресурсы: один поток (event-loop) может обслуживать тысячи подписок, без создания новых на каждую задачу. В итоге, системы становятся более отзывчивыми и масштабируемыми — идеально для микросервисов или реального времени.


Publisher: источник данных, который отправляет события

Publisher — это интерфейс для издателя, который генерирует поток данных.
Он как фабрика событий: производит элементы (любые объекты), ошибки или сигнал завершения. Publisher не знает, кто его слушает, — он просто готов отправлять (push) данные, когда они появятся.


Интерфейс прост: всего один метод subscribe(Subscriber s). Когда вы вызываете его, издатель регистрирует подписчика и начинает процесс. Но данные не льются сразу — всё под контролем (об этом ниже).

Пример простого издателя в Project Reactor (который реализует Reactive Streams):
import reactor.core.publisher.Flux; // Flux реализует Publisher

Flux<Integer> publisher = Flux.range(1, 5); // Издатель: поток от 1 до 5


Здесь publisher — источник.
Он может быть асинхронным: например, читать из сети или БД. Когда данные готовы, он отправляет их подписчику через onNext(элемент).
Если ошибка — onError(Throwable).
Если конец — onComplete().

Почему это лучше традиционных подходов? В отличие от потоков (где каждый запрос — отдельный тяжёлый объект), publisher лёгкий: он не выделяет ресурсы заранее, а реагирует на подписку. Нет блокировок: если данных нет, ничего не происходит.


#Java #middle #Reactor #Reactive_Streams_API #Processor #Subscription #Subscriber #Publisher
👍1
Subscriber: получатель, который реагирует на события

Subscriber — интерфейс для подписчика, который "слушает" издателя. Он как потребитель на конвейере: получает элементы и решает, что с ними делать.

Методы:
- onSubscribe(Subscription s): вызывается сразу после подписки. Здесь подписчик получает подписку — объект для контроля.
- onNext(T item): срабатывает на каждый элемент. Здесь обработка: логика, трансформация и т.д.
- onError(Throwable t): если ошибка — обработка исключения.
- onComplete(): поток завершён успешно.


Подписчик пассивен: не получает данные (pull), а ждёт push. Это решает callback-ад из CompletableFuture — реакции в одном месте, код чище.

Пример подписки:
publisher.subscribe(new Subscriber<Integer>() {
private Subscription subscription;

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

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

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

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


Здесь subscribe() связывает publisher и subscriber. Данные отправляются по одному, по запросу (request(1)). Это асинхронно: код после subscribe() продолжается сразу, без ожидания.

В Reactor есть BaseSubscriber для упрощения — переопределяйте только нужные методы.


Subscription: мост контроля с обратным давлением

Subscription — ключ к управлению: это объект, который подписчик получает в onSubscribe.

Методы:
- request(long n): "Дай мне n элементов". Это обратное давление (backpressure) — подписчик контролирует темп, чтобы не захлебнуться.
- cancel(): "Хватит, отпишись".


Без этого publisher мог бы зафлудить подписчика данными. Например, если источник — бесконечный поток (сенсоры), без request(n) — переполнение памяти.

В примере выше request(1) делает обработку последовательной: получил — обработал — запросил следующий. Для скорости — request(Long.MAX_VALUE) (неограниченно), но осторожно: рискуете буфером.

Это решает проблемы блокировок: вместо висящих потоков, всё в event-loop. Под нагрузкой — graceful degradation (грациозная деградация): если подписчик медленный, publisher замедляется, а не падает.


Processor: промежуточный звено для трансформаций

Processor — комбо: реализует и Publisher, и Subscriber. Он как фильтр в конвейере: принимает данные от одного издателя, обрабатывает и отправляет дальше.
Пример — в цепочках Flux: map() или filter() создают процессоры внутри.


В практике вы редко пишете свой Processor — библиотеки предоставляют готовые операторы. Но понимание помогает: весь конвейер — цепь publisher → processor → ... → subscriber.


Практические советы и подводные камни


- Всегда управляйте подпиской: без request() данные не потекут (по умолчанию unbounded — неограниченно, но лучше явно).
- Обрабатывайте ошибки: onError — ваш спасатель, чтобы не потерять исключения.
- Тестируйте с TestSubscriber (в
Reactor): симулируйте сценарии.
- Камень: если в onNext блокирующий код — сломаете асинхронность. Используйте Schedulers для offload (перенос на другой поток).


В реальном коде: в Spring WebFlux контроллеры возвращают Flux/Mono — publisher'ы, клиенты подписываются.


#Java #middle #Reactor #Reactive_Streams_API #Processor #Subscription #Subscriber
👍1
Реактивное программирование

Концепции реактивного программирования: Backpressure — что делать, если данных слишком много


Представьте: источник данных — как бурная река, а ваш подписчик — маленькая лодка. Если вода хлынет слишком быстро в лодку, она утонет.
Здесь на помощь приходит backpressure (обратное давление) — механизм, который позволяет подписчику контролировать скорость потока, говоря "дай мне столько, сколько я могу переварить". И это не просто тормоз, а умный регулятор, который предотвращает перегрузки и делает реактивные системы устойчивыми под любой нагрузкой.


Backpressure — ключевая фишка реактивного программирования, которая отличает его от традиционных подходов. В старых моделях (как потоки или Future) вы либо блокируете всё, либо тонете в очередях данных, рискуя исчерпать память или CPU. Здесь же контроль у потребителя: он решает, когда и сколько брать. Это решает проблемы callback-ада и блокировок, позволяя строить масштабируемые приложения — от мобильных до облачных кластеров.


Что такое backpressure и почему оно нужно?

Обратное давление — это способ, при котором получатель данных сигнализирует источнику: "замедлись, если я не успеваю". В реактивном мире данные передаются асинхронно, но без контроля это может привести к проблемам: если издатель генерирует 1 млн элементов в секунду, а подписчик обрабатывает только 100, буфер переполнится, и приложение упадёт с ошибкой "израсходована память" (OutOfMemoryError).

Backpressure вводит "обратную связь": подписчик запрашивает элементы порциями, а издатель выдаёт ровно столько.


Это встроено в Reactive Streams: в методе onSubscribe подписчик получает Subscription и использует request(long n) — "запроси n элементов". Если не запросить — данные не потекут. Это как шлюзы на реке: открываешь по мере нужды, избегая наводнения. В отличие от pull-модели (где вы тянете всё сразу), здесь баланс: push для динамики, но с контролем.

Пример базового использования в Flux (поток для множества элементов):
import reactor.core.publisher.Flux;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

Flux<Integer> fastPublisher = Flux.range(1, 1000); // Быстрый источник: 1000 элементов

fastPublisher.subscribe(new Subscriber<Integer>() {
private Subscription sub;

@Override
public void onSubscribe(Subscription s) {
sub = s;
sub.request(5); // Сначала запрашиваем 5 элементов
}

@Override
public void onNext(Integer item) {
System.out.println("Обработано: " + item);
// Симулируем медленную обработку: Thread.sleep(100); (но в реальности избегайте блокировок!)
sub.request(1); // После каждого — запрашиваем следующий
}

@Override
public void onError(Throwable t) { /* обработка */ }
@Override
public void onComplete() { System.out.println("Готово"); }
});

Здесь подписчик контролирует темп: запросил 5 — получил 5, обработал — запросил ещё. Если не request() — поток остановится. Это асинхронно: издатель не блокируется, а ждёт сигналов.



#Java #middle #Reactor #Reactive_Streams_API #backpressure
👍2