Как организовать подписку в RxJava

Что проверяют

Понимаете ли вы жизненный цикл подписки: кто её создаёт, на каких потоках она работает и кто её отменяет. Следом обычно спрашивают про утечки при повороте экрана и про разницу между subscribeOn и observeOn.

Как отвечать

Сразу обозначьте версию: актуальна RxJava 3 (io.reactivex.rxjava3:rxjava и rxandroid), RxJava 2 давно не поддерживается. В новом Android-коде эту нишу заняли корутины и Flow, но RxJava живёт во многих проектах.

Каркас подписки:

• тип источника: Single для одного ответа сети, Completable для операции без результата, Maybe для «значения может не быть», Observable для потока, Flowable, когда нужен backpressure;
• потоки: subscribeOn(Schedulers.io()) задаёт поток, на котором стартует источник, и действует только ближайший к источнику вызов. observeOn(AndroidSchedulers.mainThread()) переключает всё, что ниже по цепочке;
• subscribe(onSuccess, onError): возвращает Disposable. Обработчик ошибки передают всегда: без него ошибка уходит в RxJavaPlugins.onError как OnErrorNotImplementedException и роняет приложение;
• CompositeDisposable: подписки экрана складывают через add и отменяют разом. В ViewModel это делают в onCleared(), во фрагменте в onDestroyView();
• clear() или dispose(): clear() отменяет текущие подписки и оставляет контейнер рабочим, а после dispose() любая новая подписка отменяется сразу же.

Хорошо добавить, что сетевой запрос правильнее держать в ViewModel, а не в Activity: при повороте экрана он не перезапускается, а результат доходит до нового экрана через LiveData или StateFlow.

Где спотыкаются

Пишут subscribeOn несколько раз, думая, что каждый переключает поток. Работает только ближайший к источнику, а переключение ниже по цепочке делает observeOn.

Вызывают dispose() у общего CompositeDisposable в onStop, а потом добавляют новые подписки в onStart. Они отменяются в момент добавления, и экран молча перестаёт обновляться.

Не учитывают, что холодный Observable выполняется заново для каждого подписчика. Две подписки на один запрос дают два сетевых вызова, общий результат делают через share() или cache().