RxJava: основи, ReactiveX та робота з потоками даних

Що таке RxJava та ReactiveX

RxJava — це Java-реалізація специфікації ReactiveX, бібліотеки для асинхронного програмування за допомогою спостережуваних потоків (Observable). RxJava 2 була випущена в 2016 році з підтримкою Reactive Streams (Flowable) та розділенням на rx.Observable і io.reactivex.Observable. RxJava 3 (2019) — поточна мажорна версія зі зворотною сумісністю з RxJava 2.

Основна ідея RxJava — все є потоком: потік даних, потік подій, потік станів. Будь-яку асинхронну операцію можна представити як Observable, що емітить дані, помилку або сигнал завершення. Observer підписується на Observable та отримує сповіщення в реальному часі.

За даними Badoo (2024), до переходу на корутини 76% Android-додатків з топ-200 Google Play використовували RxJava для асинхронних операцій. Зараз частка знижується на користь корутин, але RxJava залишається в production-коді тисяч додатків і вважається зрілою, перевіреною технологією. ReactiveX — це кроссплатформенна специфікація, реалізована також для JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) та інших мов.

Патерн Observer в RxJava

ReactiveX розширює класичний патерн Observer двома механізмами: ланцюжок операторів та керування потоками на основі Schedulers. Observable не починає емітити дані, поки на нього не підпишеться Observer (ліниве обчислення). Це дозволяє будувати pipeline даних, який активується лише за наявності підписки.

Типи Observable: Observable, Flowable, Single, Maybe, Completable

Observable — базовий тип, що емітить 0..N елементів з onError або onComplete. Підходить для потоків даних необмеженої довжини — наприклад, подій кліків або оновлень геолокації. Observable не підтримує backpressure.

Flowable — Reactive Streams версія Observable з підтримкою backpressure. Використовується, коли джерело даних може генерувати елементи швидше, ніж Observer встигає обробляти. Flowable підтримує стратегії BACKPRESSURE_BUFFER, DROP, LATEST та ERROR.

Single, Maybe та Completable

Single емітить рівно один елемент або помилку — ідеальний для мережевих запитів. Maybe — 0 або 1 елемент, підходить для кешу, де дані можуть бути відсутні. Completable — лише onComplete або onError, без даних, зручний для операцій запису або видалення. Ці типи спрощують API, звужуючи контракт до конкретного випадку. Retrofit (популярний HTTP-клієнт для Android) підтримує всі п'ять типів RxJava безпосередньо, дозволяючи вибирати найбільш підходящий тип повернення для кожного ендпоінту без зайвого коду.

Оператори RxJava: трансформація та фільтрація потоків

Оператори — це функції, що перетворюють один Observable в інший. Ланцюжок операторів описує pipeline даних: кожен оператор приймає потік від попереднього, трансформує його та передає наступному. RxJava містить понад 200 операторів, розділених на категорії.

• map — перетворює кожен елемент (Integer → String)
• flatMap — перетворює елемент в Observable та об'єднує всі в один потік
• filter — пропускає елементи за умовою
• zip — комбінує елементи N Observable за індексом
• merge — об'єднує кілька Observable в один, зберігаючи порядок у часі
• debounce — пропускає елементи, якщо між ними менше вказаного інтервалу

flatMap — один з найпотужніших операторів RxJava. Він дозволяє виконати асинхронний запит для кожного елемента та зібрати результати в загальний потік. Наприклад, flatMap застосовується для завантаження деталей за списком ID: кожен ID → мережевий запит → об'єднання результатів. На відміну від map, який просто перетворює елемент, flatMap може емітити кілька елементів або переключатися на інший Observable, що робить його основою для побудови асинхронних pipeline.

Керування помилками через оператори

onErrorResumeNext — при помилці переключається на резервний Observable. retry — повторює підписку при помилці N разів. onErrorReturn — повертає значення за замовчуванням замість помилки. doOnError — виконує побічну дію при помилці, не змінюючи потік (логування або аналітика). Комбінування цих операторів дозволяє будувати надійні pipeline з чіткою стратегією обробки збоїв без try/catch вручну.

Schedulers: керування потоками в RxJava

Schedulers визначають, на якому потоці виконується Observable та Observer. subscribeOn задає потік для джерела, observeOn — потік для Observer та наступних операторів. Це розділення — ключова перевага RxJava: джерело на IO-потоці, обробка на computation, UI — на головному.

Основні Schedulers: Schedulers.io() — для I/O операцій (мережа, диск), необмежений пул. Schedulers.computation() — для обчислень, фіксований пул за кількістю ядер. Schedulers.newThread() — новий потік на кожне завдання. AndroidSchedulers.mainThread() — головний потік Android (RxAndroid). Також існує Schedulers.trampoline() для виконання завдань у поточному потоці з чергою FIFO, корисний для тестів.

За даними Google (2025), правильне використання Schedulers — найскладніше в RxJava для новачків. Типова помилка — виклик subscribeOn після observeOn, що не впливає на джерело. subscribeOn має бути першим у ланцюжку для джерела, observeOn — перед UI-підпискою. Правило: subscribeOn впливає лише на upstream (джерело), observeOn перемикає downstream (підписника та всі оператори після нього).

Приклади коду RxJava в Android

Розглянемо три сценарії: мережевий запит з Single, паралельні запити з zip та debounce для пошукового поля з debounce.

Мережевий запит з Single

Single ідеально підходить для Retrofit-запитів: один запит — одна відповідь. Підписка на головному потоці для оновлення UI.

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleObserver<User>() {
        @Override
        public void onSuccess(User user) { showUser(user); }
        @Override
        public void onError(Throwable e) { showError(e); }
    })

Паралельні запити з zip

zip об'єднує результати двох незалежних Single в один. Виконуються паралельно, результат — після завершення обох.

Single.zip(
    api.getProfile(),
    api.getSettings(),
    (profile, settings) -> new Dashboard(profile, settings)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(dashboard -> showDashboard(dashboard), e -> logError(e))

Debounce для пошукового поля

debounce ігнорує швидкі зміни тексту та відправляє запит лише після 400 мс паузи. distinctUntilChanged скасовує запит, якщо текст не змінився.

RxTextView.textChanges(searchView)
    .debounce(400, TimeUnit.MILLISECONDS)
    .filter(text -> text.length() >= 3)
    .distinctUntilChanged()
    .switchMap(query -> api.search(query))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> showResults(results))

RxJava vs Kotlin Coroutines: порівняння підходів

RxJava та Kotlin Coroutines вирішують одне завдання — асинхронне програмування — але принципово різними підходами. RxJava побудований на патерні Observer і є push-based: джерело надсилає дані, Observer реагує. Корутини — pull-based: код послідовно запитує дані через await.

• RxJava — реактивний, потік даних, >200 операторів, push-based, складний learning curve
• Coroutines — послідовний, suspend/await, ~40 функцій, pull-based, простий синтаксис
• RxJava — зрілий (2016), величезна екосистема, але складний learning curve
• Coroutines — сучасний (2018), кращий вибір Google для нового коду
• RxJava — backpressure з коробки через Flowable, налагоджені стратегії буферизації
• Coroutines — Flow з backpressure нещодавно, але активно розвивається JetBrains

За даними Google I/O 2024, Kotlin Coroutines — рекомендований підхід для нового асинхронного коду в Android. RxJava залишається підтримуваним для існуючих проєктів. Google надає бібліотеки-перехідники (kotlinx-coroutines-rx3) для поступової міграції. AndroidX (LiveData, Room, Paging 3) підтримує обидва підходи, дозволяючи використовувати RxJava в старих модулях і корутини в нових без конфліктів залежностей.

Стратегія міграції з RxJava на корутини

Поступовий перехід: кожен новий компонент пишеться на корутинах, старий RxJava код не чіпається. RxJava → корутини через awaitSingle() або awaitFirst(). Корутини → RxJava через future() або asFlowable(). Повна міграція займає 6–18 місяців для великих проєктів.