RxJava: суть, компоненти та реактивне програмування

Що таке RxJava?

RxJava — це реалізація бібліотеки ReactiveX (Reactive Extensions) для віртуальної машини Java. Перша версія RxJava була випущена компанією Netflix у 2013 році для управління асинхронними викликами в серверних додатках. На момент створення основною альтернативою в Java були Future та Callback — обидва підходи призводили до callback-hell та складного управління потоками. RxJava запропонувала композицію асинхронних операцій через Observable з ланцюжками функціональних операторів.

Архітектура RxJava базується на специфікації Reactive Streams — стандарті для асинхронної обробки потоків з неблокуючим backpressure. Специфікація визначає чотири інтерфейси: Publisher, Subscriber, Subscription та Processor. RxJava 2+ повністю реалізує Reactive Streams через тип Flowable, дотримуючись контрактів backpressure на відміну від RxJava 1. Observable в RxJava 2 не підтримує backpressure — він призначений для потоків з невеликою кількістю подій або UI-подій.

Згідно з опитуванням JetBrains, 2025, RxJava входить до топ-3 бібліотек для Android-розробки. Основні сценарії використання: обробка мережевих запитів через Retrofit (інтегрований з RxJava через CallAdapter), робота з Room (реактивні запити повертають Flowable або Maybe), анімації та UI-події через RxBinding, та дебаунс-пошук при введенні тексту. Всі ці сценарії об'єднує однотипний ланцюжок: джерело (Observable) → трансформація (оператори) → підписка (subscribe).

Історія версій RxJava

RxJava 1 (2013) заклав концепцію Observable та операторів, але страждав від проблем із backpressure — у швидких потоках дані накопичувалися в пам'яті, викликаючи OutOfMemoryError. RxJava 2 (2016) виправив архітектуру, розділивши Observable (без backpressure) та Flowable (з backpressure). RxJava 3 (2020) додав підтримку Java 8 Stream API, додаткові оператори та покращену продуктивність підписки. На поточний момент RxJava 3 — рекомендована версія для нових проектів.

Типи реактивних потоків у RxJava

RxJava надає п'ять основних типів реактивних джерел, кожен з яких орієнтований на певний сценарій. Observable та Flowable випускають безліч значень, Single — одне значення або помилку, Completable — лише факт завершення без даних, Maybe — одне значення, нуль або помилку. Вибір правильного типу скорочує кількість коду та робить ланцюжок самодокументованим.

Flowable — найбільш гнучкий тип для роботи з великими потоками даних. Він реалізує Reactive Streams Publisher з підтримкою backpressure: споживач може запросити певну кількість елементів через Subscription.request(n). Це запобігає переповненню буфера при невідповідності швидкостей виробника та споживача. Якщо backpressure не критичний — використовуйте Observable, він має менше накладних витрат через відсутність механізму request.

Single — оптимальний вибір для HTTP-запитів. Retrofit 2 з RxJava CallAdapter повертає Single<ResponseBody> для кожного запиту. Single гарантує рівно один виклик onSuccess або onError, що відповідає семантиці HTTP-запиту — одна відповідь або одна помилка. Completable використовується для операцій запису, що не повертають даних: insert, update, delete. Maybe зручний при перевірці кешу — може повернути значення, а може не повернути.

// Приклад використання Single для HTTP-запиту
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Підписка з обробкою на головному потоці
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Оператори трансформації та управління потоками

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

Оператори трансформації

map — базовий оператор, що перетворює кожне значення через функцію. flatMap приймає функцію, яка повертає Observable для кожного елемента, та розгортає результат в єдиний потік. switchMap схожий на flatMap, але при надходженні нового елемента відписується від попереднього Observable. concatMap зберігає порядок елементів — на відміну від flatMap, він послідовно підписується на кожен вкладений Observable.

// Парсинг JSON з трансформацією та фільтрацією
apiService.getUsers()
    .flatMap { users ->
        Observable.fromIterable(users)
    }
    .filter { user ->
        user.age >= 18
    }
    .map { user ->
        UserDto(user.name, user.age)
    }
    .toList()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ adapter.submitList(it) },
               { Log.e("Помилка", it.message) })

Комбінування потоків — область, де RxJava особливо сильний. zip об'єднує елементи з кількох Observable попарно за індексом: перший з першим, другий з другим. combineLatest випускає нове значення при зміні будь-якого з потоків, комбінуючи останні значення всіх потоків. merge об'єднує кілька Observable в один, зберігаючи порядок надходження подій. concat послідовно підписується на кожен Observable та передає всі його події, перш ніж перейти до наступного.

Управління часом включає debounce (очікування паузи в потоці перед відправкою), throttleFirst (пропуск першої події, ігнорування інших протягом вікна), timeout (помилка, якщо подія не надійшла протягом інтервалу). Дебаунс-пошук при введенні тексту — найпоширеніший сценарій: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() запобігає зайвим запитам при швидкому наборі.

Schedulers та багатопотоковість

Scheduler у RxJava — це абстракція над пулом потоків. Бібліотека надає п'ять вбудованих Scheduler: Schedulers.io() для I/O-операцій (мережа, файли), Schedulers.computation() для CPU-інтенсивних завдань, Schedulers.newThread() для кожного нового потоку, Schedulers.single() для однопотокового виконання та Schedulers.trampoline() для негайного виконання в поточному потоці.

subscribeOn та observeOn

subscribeOn визначає, на якому Scheduler виконується джерело Observable. Якщо в ланцюжку кілька subscribeOn — пріоритет у найближчого до джерела. observeOn перемикає downstream на вказаний Scheduler — кожне використання observeOn змінює потік для наступних операторів. Типовий Android-патерн: subscribeOn(Schedulers.io()) для роботи з мережею, observeOn(AndroidSchedulers.mainThread()) для оновлення UI.

// Багатопотокова обробка з перемиканням контексту
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // БД на io
    .map(items -> processItems(items))     // трансформація на io
    .observeOn(Schedulers.computation())    // перемикаємо на computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() — Scheduler з бібліотеки RxAndroid, який виконує код на головному потоці Android. Він обов'язковий для будь-яких оновлень UI в реактивному ланцюжку. Бібліотека використовує Handler всередині та гарантує виконання в UI-потоці навіть при високому навантаженні. Для фонових операцій Schedulers.io() підтримує необмежений пул потоків та підходить для будь-яких блокуючих операцій. Schedulers.computation() використовує фіксований пул, рівний кількості ядер процесора.

RxJava в Android: практичне застосування

RxJava в Android використовується для трьох основних сценаріїв: реактивні запити до Room, інтеграція з Retrofit та реактивне зв'язування UI через RxBinding. Для кожного сценарію характерний свій набір типів: Room повертає Flowable для спостережуваних запитів, Retrofit — Single для HTTP-запитів, RxBinding — Observable для UI-подій.

Room + RxJava

Room — бібліотека персистентності даних від Google. Починаючи з Room 2.1, база даних підтримує реактивні типи, що повертаються: Flowable та Observable. При зміні будь-якого запису в таблиці Room автоматично надсилає нове значення в потік. Розробник підписується на Flowable в ViewModel та отримує актуальні дані без ручних запитів при кожній зміні.

// Room DAO з реактивним запитом
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun getUserById(@Param("id") userId: Int): Flowable<User>

    @Insert
    fun insertUser(user: User): Completable
}

// ViewModel — композиція Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Патерн MVVM + RxJava будується на тому, що ViewModel не має посилань на View. ViewModel публікує реактивні джерела (Flowable, LiveData через Transformations), а Activity або Fragment підписуються на них. Це дає тестованість: ViewModel тестується без UI, підміняючи Scheduler через RxJavaPlugins.setComputationScheduler. CompositeDisposable у ViewModel керує життєвим циклом підписок — при onCleared() всі підписки скасовуються.

RxJava vs Kotlin Flow

Kotlin Flow — нативна реалізація холодних потоків у Kotlin, вбудована в корутини та представлена в Kotlin 1.3. Flow вирішує ті самі завдання, що й RxJava, але з фундаментальними відмінностями: вбудована підтримка корутин (suspend-функції), скасування через coroutine cancellation та відсутність проблем із backpressure — Flow використовує suspend замість буферизації. Flow є частиною стандартної бібліотеки Kotlin, не вимагаючи додаткових залежностей.

RxJava залишається кращим вибором для проектів на Java, проектів з підтримкою Java 7-8 та існуючих кодових баз на RxJava. Екосистема RxJava значно багатша: >400 операторів проти ~50 у Flow, інтеграція з Retrofit через вбудований CallAdapter, підтримка backpressure через Flowable та наявність RxBinding, RxPermissions, RxLocation для Android. Kotlin Flow стрімко наздоганяє, але гнучкість RxJava в складних сценаріях комбінування потоків поки вища.