Kotlin Coroutines против RxJava: сетевые запросы и фоновые задачи

Почему этот выбор важен для production‑сетевых задач

Сетевые запросы и фоновая работа в реальном Android‑приложении — это не один API‑вызов. Сюда входят вход в систему и обновление токена, экраны, которые могут поворачиваться во время загрузки, синхронизация после ухода пользователя со страницы, загрузка фото и периодические задачи, которые нельзя выполнять в ущерб батарее.

Багаи, которые больше всего бьют по пользователям, обычно не про синтаксис. Они появляются, когда асинхронная работа живёт дольше UI (утечки), когда отмена останавливает интерфейс, но не сам запрос (трафик тратится и крутилка не исчезает), когда ретраи умножают запросы (лимиты, баны), или когда разные слои по‑разному обрабатывают ошибки и никто не может предсказать, что увидит пользователь.

Выбор между Kotlin Coroutines и RxJava влияет на повседневную надёжность:

• Как вы моделируете работу (одноразовые вызовы против потоков)
• Как распространяется отмена
• Как ошибки представлены и доставляются в UI
• Как вы управляете потоками для сети, диска и UI
• Насколько легко тестировать тайминги, ретраи и граничные случаи

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

Основные ментальные модели: suspend‑вызовы, потоки и Flow

Главное различие между Kotlin Coroutines и RxJava — это форма работы, которую вы моделируете.

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

RxJava изначально спрашивает, имеете ли вы дело с одним значением или множеством значений во времени. Single — это одноразовый результат (успех или ошибка). Observable (или Flowable) — это поток, который может эмитить много значений, потом завершиться или провалиться. Это подходит для действительно событийных фич: изменения текста, сообщения веб‑сокета или опрос.

Flow — это дружественный к корутинам способ представить поток. Можно думать о нём как о «потоковой версии» корутин с структурированной отменой и прямой интеграцией с suspend‑API.

Короткое правило:

• Используйте suspend для одного запроса и одного ответа.
• Используйте Flow для значений, которые меняются со временем.
• Используйте RxJava, если приложение уже сильно опирается на операторы и сложную композицию потоков.

По мере роста функциональности читаемость обычно ломается первой, когда вы насильно ставите потоковую модель на одноразовый вызов или пытаетесь обращаться с непрерывными событиями как с единым возвращаемым значением. Сначала подгоняйте абстракцию под реальность, затем вырабатывайте соглашения.

Отмена на практике (с короткими примерами кода)

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

В Kotlin Coroutines отмена встроена в модель. Job представляет работу, и со структурированной конкуренцией обычно не нужно передавать джобы вокруг. Вы запускаете работу внутри скоупа (например, viewModelScope). Когда скоуп отменяется, всё внутри тоже отменяется.

class ProfileViewModel(
    private val api: Api
) : ViewModel() {

    fun loadProfile() = viewModelScope.launch {
        // If the ViewModel is cleared, this coroutine is cancelled,
        // and so is the in-flight network call (if the client supports it).
        val profile = api.getProfile() // suspend
        // update UI state here
    }
}

Два производственных момента важны:

• Вызывайте сетевые suspend‑вызовы через клиент, поддерживающий отмену. Иначе корутина остановится, а HTTP‑запрос может продолжить выполняться.
• Используйте withTimeout (или withTimeoutOrNull) для запросов, которые не должны висеть вечно.

RxJava использует явное освобождение ресурсов (disposable). Вы храните Disposable для каждой подписки или собираете их в CompositeDisposable. Когда экран уходит, вы делаете dispose, и цепочка должна остановиться.

class ProfilePresenter(private val api: ApiRx) {
    private val bag = CompositeDisposable()

    fun attach() {
        bag += api.getProfile()
            .subscribe(
                { profile -> /* render */ },
                { error -> /* show error */ }
            )
    }

    fun detach() {
        bag.clear() // cancels in-flight work if upstream supports cancellation
    }
}

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

Обработка ошибок, которая остаётся понятной

Большое отличие между Kotlin Coroutines и RxJava — это путь распространения ошибок. В корутинах сбои выглядят как обычный код: suspend‑вызов бросает, и вызывающий решает, что с этим делать. Rx проталкивает ошибки через поток — это мощно, но легко спрятать проблему, если не быть осторожным.

Используйте исключения для неожиданных сбоев (таймауты, 500‑ые, ошибки парсинга). Моделируйте ошибки как данные, когда UI ожидает конкретный ответ (неверный пароль, «почта уже занята») и вы хотите, чтобы это было частью доменной модели.

Простой корутинный паттерн сохраняет стек и остаётся читаемым:

suspend fun loadProfile(): Profile = try {
    api.getProfile() // may throw
} catch (e: IOException) {
    throw NetworkException("No connection", e)
}

runCatching и Result полезны, когда вы действительно хотите вернуть успех или ошибку без броска:

suspend fun loadProfileResult(): Result<Profile> =
    runCatching { api.getProfile() }

Будьте аккуратны с getOrNull(), если вы не обрабатываете ошибку дополнительно. Это может тихо превратить реальные баги в экраны «пустого состояния».

В RxJava держите путь ошибок явным. Используйте onErrorReturn только для безопасных фоллбеков. Предпочитайте onErrorResumeNext, когда нужно переключиться на другой источник (например, кэшу). Для ретраев ограничивайте правила через retryWhen, чтобы не ретраить на «неверный пароль».

Набор привычек, который предотвращает заглатывание ошибок:

• Логируйте или репортите ошибку один раз, близко к месту, где есть контекст.
• Сохраняйте оригинальное исключение как cause при обёртке.
• Избегайте универсальных фолбеков, которые превращают любую ошибку в значение по умолчанию.
• Делайте ошибки, отображаемые пользователю, типизированной моделью, а не строкой.

Основы потоков: Dispatchers vs Schedulers

Многие баги асинхронности сводятся к потокам: тяжёлая работа на главном потоке или попытка трогать UI из фонового потока. Kotlin Coroutines и RxJava отличаются главным образом способом выражения переключений потоков.

В корутинах вы часто стартуете на главном потоке для UI, затем прыгаете на фоновый диспетчер для тяжёлых частей. Обычные варианты:

• Dispatchers.Main для обновлений UI
• Dispatchers.IO для блокирующего I/O (сеть, диск)
• Dispatchers.Default для CPU‑работы (парсинг JSON, сортировка, шифрование)

Простой паттерн: получить данные, распарсить вне main, затем отрендерить.

viewModelScope.launch(Dispatchers.Main) {
    val json = withContext(Dispatchers.IO) { api.fetchProfileJson() }
    val profile = withContext(Dispatchers.Default) { parseProfile(json) }
    _uiState.value = UiState.Content(profile)
}

RxJava выражает «где выполняется работа» через subscribeOn и «где наблюдаются результаты» через observeOn. Частая неожиданность — ожидание, что observeOn повлияет на upstream; это не так. subscribeOn задаёт поток для источника и операторов выше, а каждый observeOn переключает поток с той точки и дальше.

api.fetchProfileJson()
    .subscribeOn(Schedulers.io())
    .map { json -> parseProfile(json) } // still on io unless you change it
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { profile -> render(profile) },
        { error -> showError(error) }
    )

Правило, которое избегает сюрпризов: держите UI‑работу в одном месте. В корутинах назначайте или коллекционируйте UI‑стейт на Dispatchers.Main. В RxJava ставьте финальный observeOn(main) прямо перед рендером и не разбрасывайте observeOn по коду без необходимости.

Если экран подтормачивает, сначала вынесите парсинг и мэппинг с главного потока. Это одиночное изменение решает много реальных проблем.

Таймауты, ретраи и параллельная работа для сетевых вызовов

Счастливый путь редко вызывает проблемы. Беда приходит от висящих вызовов, ретраев, которые усугубляют ситуацию, или «параллельной» работы, которая на деле не параллельна. Эти паттерны часто определяют, почему команда выбирает Kotlin Coroutines или RxJava.

Таймауты, которые быстро падают

В корутинах вы можете поставить жёсткую границу вокруг любого suspend‑вызова. Держите таймаут близко к месту вызова, чтобы показать правильное сообщение в UI.

val user = withTimeout(5_000) {
    api.getUser() // suspend
}

В RxJava вы прикрепляете оператор timeout к потоку. Это удобно, когда поведение таймаута является частью общего пайплайна.

Ретраи без вреда

Ретрайте только когда это безопасно. Простое правило: ретрайте идемпотентные запросы (GET) более свободно, чем запросы с сайд‑эффектами (например, создание заказа). В любом случае ограничьте количество попыток и добавьте задержку или джиттер.

Хорошие guardrails по умолчанию:

• Ретрайте на сетевые таймауты и временные ошибки сервера.
• Не ретрайте на ошибки валидации (400) или проблемы авторизации.
• Ограничьте число ретраев (обычно 2–3) и логируйте финальный провал.
• Используйте backoff, чтобы не бить по серверу.

В RxJava retryWhen позволяет выразить «ретрай только для этих ошибок, с такой задержкой». В корутинах Flow имеет retry и retryWhen, а для обычных suspend‑функций часто достаточно небольшого цикла с delay.

Параллельные вызовы без запутанного кода

Корутины делают параллельную работу прямой: запусти два запроса и дождись обоих.

coroutineScope {
    val profile = async { api.getProfile() }
    val feed = async { api.getFeed() }
    profile.await() to feed.await()
}

RxJava хороша, когда комбинирование нескольких источников — суть цепочки. zip обычно используется, чтобы дождаться обоих, а merge — когда вы хотите результаты по мере их прихода.

Для больших или быстрых потоков всё ещё важен backpressure. Flowable в RxJava имеет зрелые инструменты для backpressure. Coroutines Flow справляется во многих случаях, но иногда требуется буферизация или политика сброса, если события опережают UI или записи в базу.

Взаимодействие и стратегии миграции (смешанные кодовые базы)

Большинство команд не меняют всё мгновенно. Практичная миграция между Kotlin Coroutines и RxJava сохраняет приложение стабильным и позволяет двигаться модуль за модулем.

Обернуть Rx API в suspend‑функцию

Если у вас есть Single<T> или Completable, оберните их с поддержкой отмены, чтобы отмена корутины диспоузила Rx‑подписку.

suspend fun <T : Any> Single<T>.awaitCancellable(): T =
  suspendCancellableCoroutine { cont ->
    val d = subscribe(
      { value -> cont.resume(value) {} },
      { error -> cont.resumeWithException(error) }
    )
    cont.invokeOnCancellation { d.dispose() }
  }

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

Открыть корутинный код для Rx‑вызывателей

Во время миграции некоторые слои всё ещё будут ожидать Rx‑типы. Обёртывайте suspend‑работу в Single.fromCallable, блокируя только на фоновом потоке.

fun loadProfileRx(api: Api): Single<Profile> =
  Single.fromCallable {
    runBlocking { api.loadProfile() } // ensure subscribeOn(Schedulers.io())
  }

Держите эту границу маленькой и документированной. Для нового кода предпочитайте вызывать suspend API напрямую из корутинного скоупа.

Где подходит Flow, а где нет

Flow может заменить многие случаи Observable: состояние UI, обновления базы, потоки для пагинации. Он может быть менее прямым решением, если вы сильно полагаетесь на hot‑потоки, subjects, тонкую настройку backpressure или большой набор кастомных операторов, с которыми команда уже знакома.

Стратегия миграции, снижающая путаницу:

• Сначала конвертируйте leaf‑модули (сеть, хранение) в suspend API.
• Добавьте небольшие адаптеры на границах модулей (Rx→suspend, suspend→Rx).
• Заменяйте Rx‑потоки на Flow, только когда вы контролируете потребителей.
• Держите один асинхронный стиль на область функциональности.
• Удаляйте адаптеры сразу после миграции последнего потребителя.

Паттерны тестирования, которые вы действительно будете использовать

Проблемы с таймингом и отменой — то место, где прячутся асинхронные баги. Хорошие асинхронные тесты делают время детерминированным и результаты простыми для ассертов. Здесь Kotlin Coroutines и RxJava ощущаются по‑разному, хотя оба стека можно тестировать хорошо.

Корутины: runTest, TestDispatcher и контроль времени

Для корутин предпочитайте runTest с тест‑диспетчером, чтобы тест не зависел от реальных потоков или задержек. Виртуальное время позволяет триггерить таймауты, ретраи и debounce‑окна без sleep.

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun `emits Loading then Success`() = runTest {
  val dispatcher = StandardTestDispatcher(testScheduler)
  val repo = Repo(api = fakeApi, io = dispatcher)

  val states = mutableListOf<UiState>()
  val job = launch(dispatcher) { repo.loadProfile().toList(states) }

  testScheduler.runCurrent() // run queued work
  assert(states.first() is UiState.Loading)

  testScheduler.advanceTimeBy(1_000) // trigger delay/retry windows
  testScheduler.runCurrent()
  assert(states.last() is UiState.Success)

  job.cancel()
}

Чтобы тестировать отмену, отменяйте собирающий Job (или родительский скоуп) и проверяйте, что фейковый API остановился или что больше нет эмиссий.

RxJava: TestScheduler, TestObserver, детерминированное время

Rx‑тесты обычно комбинируют TestScheduler для времени и TestObserver для утверждений.

@Test
fun `disposes on cancel and stops emissions`() {
  val scheduler = TestScheduler()
  val observer = TestObserver<UiState>()

  val d = repo.loadProfileRx(scheduler)
    .subscribeWith(observer)

  scheduler.triggerActions()
  observer.assertValueAt(0) { it is UiState.Loading }

  d.dispose()
  scheduler.advanceTimeBy(1, TimeUnit.SECONDS)
  observer.assertValueCount(1) // no more events after dispose
}

При тестировании путей ошибок в любом стиле фокусируйтесь на маппинге, а не на типе исключения. Проверяйте ожидаемый стейт UI после 401, таймаута или некорректного ответа.

Небольшой набор проверок покрывает большинство регрессий:

• Loading и конечные состояния (Success, Empty, Error)
• Очистка при отмене (job cancelled, disposable disposed)
• Маппинг ошибок (коды сервера → пользовательские сообщения)
• Отсутствие дублирующих эмиссий после ретраев
• Логика, зависящая от времени, с виртуальным временем, а не реальными sleep

Частые ошибки, приводящие к проблемам в проде

Большинство прод‑проблем не из‑за выбора Coroutines vs RxJava. Они возникают из нескольких привычек, которые заставляют работу выполняться дольше чем вы думаете, запускаться дважды или трогать UI из неправильного потока.

Одна из утечек — запуск работы в неправильном скоупе. Если вы стартуете сетевой вызов из скоупа, который живёт дольше экрана (или вы создаёте собственный скоуп и никогда не отменяете его), запрос может завершиться после ухода пользователя и всё равно попытаться обновить стейт. В корутинах это часто выглядит как использование долгоживущего скоупа по умолчанию. В RxJava это обычно забытый dispose.

Другой классический случай — «fire and forget». Глобальные скоупы и забытые Disposables кажутся безопасными, пока работы не накопятся. Экран чата, который обновляется на каждом onResume, легко может породить несколько параллельных задач после нескольких навигаций, каждая из которых держит память и борется за сеть.

Ретраи тоже легко сделать неправильно. Неограниченные ретраи или ретраи без задержки могут заспамить бэкенд и разрядить батарею. Особенно опасны они при постоянных ошибках, например 401 после выхода из аккаунта. Делайте ретраи условными, добавляйте backoff и прекращайте, когда ошибка не восстанавливаемая.

Ошибки с потоками приводят к крашам, которые трудно воспроизвести. Вы можете парсить JSON на main‑потоке или обновлять UI из фонового потока в зависимости от того, где вы поставили диспетчер или scheduler.

Быстрые проверки, ловящие большинство проблем:

• Привязывайте работу к объекту жизненного цикла и отменяйте её при окончании владельца.
• Делайте очистку очевидной: отменяйте Jobs или очищайте Disposables в одном месте.
• Жёстко ограничьте ретраи (кол‑во, задержка, какие ошибки подходят).
• В ревью кодов требуйте одно правило для обновлений UI (только main‑поток).
• Рассматривайте фоновые синки как систему с ограничениями, а не как случайный вызов функции.

Если вы выпускаете Android‑приложения из генерируемого Kotlin‑кода (например, из AppMaster), те же ловушки остаются в силе. Всё равно нужны понятные соглашения по скоупам, отмене, лимитам ретраев и правилам потоков.

Короткий чеклист для выбора Coroutines, RxJava или обоих

Начните с формы работы. Большинство сетевых вызовов — одноразовые, но в приложениях есть и сигналы, которые идут постоянно: состояние соединения, авторизация или live‑обновления. Неподходящая абстракция на раннем этапе обычно проявляется позже в виде запутанной отмены и трудночитаемых путей обработки ошибок.

Простой способ принять решение (и объяснить его команде):

• Одноразовый запрос (логин, получить профиль): предпочитайте suspend‑функцию.
• Непрерывный поток (события, обновления БД): предпочитайте Flow или Rx Observable.
• Отмена по lifecycle: корутины в viewModelScope или lifecycleScope часто проще, чем ручные Disposables.
• Сильная зависимость от операторов и backpressure: RxJava может оставаться лучшим выбором, особенно в старых кодовых базах.
• Сложные ретраи и маппинг ошибок: выбирайте подход, который команда сможет держать читаемым.

Практическое правило: если один экран делает один запрос и рендерит результат — корутины держат код близким к обычной функции. Если вы строите pipeline из множества событий (тайпинг, debounce, отмена предыдущих запросов, комбинирование фильтров), RxJava или Flow обычно ощущаются естественнее.

Последовательность важнее совершенства. Две хорошие практики, применённые повсеместно, легче поддерживать чем пять «лучших» паттернов, используемых непоследовательно.

Пример сценария: логин, загрузка профиля и фоновая синхронизация

Типичный продовый поток: пользователь нажимает Login, вы вызываете auth‑эндпоинт, затем получаете профиль для домашнего экрана и, наконец, запускаете фоновую синхронизацию. Здесь разница между Kotlin Coroutines и RxJava проявляется в повседневной поддержке.

Версия на корутинах (последовательно + отменяемо)

Корутины хорошо выражают «сделать это, потом то». Если пользователь закроет экран, отмена скоупа остановит незавершённую работу.

suspend fun loginAndLoadProfile(): Result<Profile> = runCatching {
  val token = api.login(email, password) // suspend
  val profile = api.profile("Bearer $token")
  syncManager.startSyncInBackground(token) // fire-and-forget
  profile
}.recoverCatching { e ->
  throw when (e) {
    is HttpException -> when (e.code()) {
      401 -> AuthExpiredException()
      in 500..599 -> ServerDownException()
      else -> e
    }
    is IOException -> NoNetworkException()
    else -> e
  }
}

// UI layer
val job = viewModelScope.launch { loginAndLoadProfile() }
override fun onCleared() { job.cancel() }

Версия на RxJava (цепочка + disposal)

В RxJava тот же поток выглядит как цепочка. Отмена — это disposal, обычно через CompositeDisposable.

val d = api.login(email, password)
  .flatMap { token -> api.profile("Bearer $token").map { it to token } }
  .doOnSuccess { (_, token) -> syncManager.startSyncInBackground(token) }
  .onErrorResumeNext { e: Throwable ->
    Single.error(
      when (e) {
        is HttpException -> if (e.code() == 401) AuthExpiredException() else e
        is IOException -> NoNetworkException()
        else -> e
      }
    )
  }
  .subscribe({ (profile, _) -> show(profile) }, { showError(it) })

compositeDisposable.add(d)
override fun onCleared() { compositeDisposable.clear() }

Минимальный набор тестов в этом сценарии должен покрывать три исхода: успех, корректный маппинг ошибок (401, 500‑ые, нет сети) и отмену/dispose.

Следующие шаги: выберите соглашения и держитесь их

Команды обычно вляпываются, потому что соглашения варьируются в разных фичах, а не потому что Kotlin Coroutines vs RxJava «плох». Короткая записка с принятым решением (даже на одной странице) экономит время на ревью и делает поведение предсказуемым.

Начните с чёткого разделения: одноразовая работа (один сетевой вызов, который возвращается один раз) против потоков (обновления во времени, websocket, локация, изменения БД). Решите дефолт для каждого и определите, когда допустимы исключения.

Затем добавьте небольшой набор общих хелперов, чтобы каждая фича вела себя одинаково при проблемах сети:

• Одно место для маппинга ошибок (HTTP‑коды, таймауты, офлайн) в прикладные ошибки, понятные UI
• Значения таймаутов по умолчанию для сетевых вызовов с возможностью переопределения для долгих операций
• Политика ретраев, описывающая, что безопасно ретраить (например, GET vs POST)
• Правило отмены: что останавливается при уходе пользователя, а что может продолжиться
• Правила логирования, помогающие поддержке без утечек приватных данных

Конвенции по тестам не менее важны. Договоритесь о стандарте, чтобы тесты не зависели от реального времени или реальных потоков. Для корутин это обычно значит тест‑диспетчер и структурированные скоупы. Для RxJava — TestScheduler и явный dispose. В любом случае старайтесь делать быстрые детерминированные тесты без sleep.

Если вы хотите двигаться быстрее в целом, AppMaster (appmaster.io) — один из вариантов для генерации бэкенд‑API и Kotlin‑мобильных приложений без написания всего вручную. Даже с генерируемым кодом те же production‑соглашения по отмене, ошибкам, ретраям и тестированию сохраняют поведение сетевых операций предсказуемым.