RxJava

RxJava - реактивная библиотека для композиции асинхронных потоков и событий. Для нового Android-кода она чаще считается legacy-решением: обычно предпочитают Coroutines и Flow. Однако RxJava всё ещё широко встречается в production-приложениях и при миграции старого кода.

В Rx-цепочке есть три основных элемента:

• источник, например Observable, Single или Flowable;
• операторы, преобразующие, объединяющие, планирующие или наблюдающие emissions;
• подписка, представленная Disposable.

Если оператор не задаёт иное поведение, ошибка завершает поток. Освобождение подписки останавливает доставку downstream, но отмена самой операции зависит от поддержки отмены источником.

Основные реактивные типы

Observable и Flowable

Observable<T> отправляет от нуля до нескольких элементов и не реализует backpressure-протокол Reactive Streams. Он подходит для UI-событий и обычных потоков, где producer не может перегрузить consumer либо sampling и отбрасывание данных заданы явно.

Flowable<T> предназначен для потоков, в которых producer может быть быстрее downstream. Backpressure позволяет downstream запрашивать ограниченное количество элементов, а не принимать неограниченный push-поток. Он полезен для больших или высокочастотных потоков, но не обязателен для каждого запроса к базе данных или сети.

При преобразовании источника без backpressure через toFlowable(strategy) выбирайте стратегию по контракту данных:

• BUFFER сохраняет ожидающие элементы, но неограниченный буфер может исчерпать память;
• DROP отбрасывает элементы, которые не удаётся вовремя доставить;
• LATEST сохраняет только самый новый ожидающий элемент;
• ERROR выдаёт MissingBackpressureException, если downstream не успевает запросить данные;
• MISSING не задаёт стратегию на границе, поэтому backpressure должен обработать следующий оператор.

Стратегия - не переключатель производительности. Отбросить платёжное событие недопустимо, а промежуточные показания датчика иногда можно потерять.

Single, Maybe и Completable

• Single<T> отправляет ровно одно значение или ошибку.
• Maybe<T> отправляет одно значение, завершается без значения либо сообщает об ошибке.
• Completable не отправляет значение, а только завершается успешно или с ошибкой.

fun loadUser(id: String): Single<User>
fun findCachedUser(id: String): Maybe<User>
fun updateUser(user: User): Completable

Выбирайте наиболее узкий тип, точно выражающий контракт. Возможно отсутствующая строка базы данных обычно представляется Maybe, а сохранение, для которого важны только успех или ошибка, - Completable.

Преобразование внутренних источников

map и flatMap

map выполняет синхронное преобразование один к одному и не создаёт новый реактивный источник.

usersObservable.map { user -> user.name }

flatMap преобразует каждый элемент во внутренний источник и объединяет его emissions с внешним потоком. Он подходит для зависимой асинхронной работы.

userIdsObservable
    .flatMap { id -> api.getUser(id).toObservable() }

flatMap, concatMap и switchMap

Эти операторы отличаются параллельностью, порядком и отменой:

• flatMap может одновременно подписываться на несколько внутренних источников и не сохраняет порядок их emissions;
• concatMap ждёт завершения каждого внутреннего источника перед подпиской на следующий и сохраняет исходный порядок;
• concatMapEager подписывается заранее, но буферизует результаты, чтобы отправить их в исходном порядке;
• switchMap освобождает предыдущую внутреннюю подписку при появлении нового элемента и оставляет активным только последний источник.

switchMap удобен для поисковой строки, но освобождение подписки не откатывает транзакцию. Отключённый сетевой запрос всё ещё может дойти до сервера, поэтому корректность операций записи не должна зависеть от switchMap.

queryChanges
    .debounce(300, TimeUnit.MILLISECONDS)
    .distinctUntilChanged()
    .switchMap { query -> api.search(query).toObservable() }

Планирование выполнения

subscribeOn и observeOn

subscribeOn(scheduler) планирует подписку на upstream-источник. Его положение в простой цепочке не меняет источник, на который он действует. При нескольких subscribeOn место начальной подписки обычно определяет оператор, ближайший к источнику; лишние переключения затрудняют понимание цепочки.

observeOn(scheduler) переносит downstream-уведомления на указанный scheduler от места вызова до следующего observeOn. Он также создаёт асинхронную границу и буферизацию, поэтому ненужные переключения имеют цену.

api.loadUser()
    .subscribeOn(Schedulers.io())
    .map { it.toUiModel() }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { uiModel -> render(uiModel) },
        { error -> showError(error) },
    )

Schedulers.io() предназначен для блокирующего I/O и может увеличивать пул потоков. CPU-intensive работа должна выполняться на Schedulers.computation(), а обращение к UI - на AndroidSchedulers.mainThread().

Создание источников

just, fromIterable и defer

just(value) захватывает и отправляет переданное значение. Observable.just(list) отправляет один элемент - сам список. fromIterable(list) отправляет каждый элемент списка отдельно.

Observable.just(listOf("A", "B"))          // Observable<List<String>>
Observable.fromIterable(listOf("A", "B")) // Observable<String>

defer создаёт новый источник для каждого subscriber в момент подписки. Используйте его, если значение или решение нужно вычислять лениво, а не захватывать при сборке цепочки.

val currentTime = Observable.defer {
    Observable.just(System.currentTimeMillis())
}

Объединение потоков

• merge передаёт emissions всех источников по мере поступления без гарантии порядка между источниками;
• concat подписывается на источники последовательно и ждёт завершения каждого;
• zip объединяет первый элемент каждого источника, затем второй и так далее;
• combineLatest начинает работу после первого значения каждого источника, а затем реагирует на изменение любого из них;
• withLatestFrom отправляет результат только при emission основного источника, добавляя последние значения остальных;
• switchOnNext принимает поток потоков и передаёт данные только из последнего внутреннего источника.

Observable.combineLatest(user, settings) { userValue, settingsValue ->
    UiModel(userValue, settingsValue)
}

Учитывайте источники, которые никогда не завершаются: concat может не перейти к следующему источнику, а zip - бесконечно ждать следующую пару.

Side effects, ошибки и освобождение подписок

doOnNext, doOnError, doOnComplete, doOnSubscribe, doOnDispose и doFinally наблюдают события жизненного цикла, не меняя обычные emissions. Они полезны для логирования, метрик, трассировки и очистки ресурсов. Не прячьте изменения бизнес-состояния в callback-функциях doOn...: для них лучше использовать явные операторы или subscribers.

repository.observeItems()
    .doOnNext { items -> logger.log("items=${items.size}") }
    .doOnError(logger::log)
    .doFinally { progress.hide() }

Всегда задавайте обработку ошибок. Вариант subscribe с одним аргументом не содержит прикладного error handler, поэтому необработанная ошибка может попасть в RxJavaPlugins.onError. Используйте onErrorReturn, onErrorResumeNext и retry только при соответствии их семантики операции. Бесконечные retry-циклы могут усилить сбой сервиса или повторить неидемпотентную запись.

В Android привязывайте подписки к явному владельцу. Обычно CompositeDisposable очищают при уничтожении соответствующего экрана, presenter или компонента. Сохранённая подписка, захватившая Activity, Fragment или View, может вызвать утечку либо позднее обновление уничтоженного UI.

Миграция на Coroutines и Flow

Мигрируйте по границам, а не переписывайте сразу все цепочки. Сначала зафиксируйте поведение тестами, включая порядок, отмену, retry и отсутствие результата. Затем замените один API или одну фичу, временно оставив adapters на границе двух моделей.

Типичные концептуальные соответствия:

• Single<T> -> suspend () -> T;
• Maybe<T> -> suspend () -> T?, если пустое завершение означает только отсутствие значения;
• Completable -> suspend () -> Unit;
• Observable<T> / Flowable<T> -> Flow<T> с явно выбранной политикой буферизации;
• switchMap -> flatMapLatest, concatMap -> flatMapConcat.

Эти типы не эквивалентны механически. Rx disposal и отмена coroutine распространяются по-разному, Flow по умолчанию cold, а backpressure-стратегии не имеют однозначного соответствия в буферизации Flow. На каждом bridge проверяйте ownership и отмену.

Связанные темы: основы Flow, операторы Flow и Coroutine Scopes & Cancellation.

Источники: проект и документация RxJava, справочник операторов ReactiveX.