Covers: reactive basics, Observable types, Schedulers, operators, Subjects, hot vs cold, backpressure, disposal, error handling, migration to Flow. (Still asked often for legacy/large codebases.)
Answer
Reactive programming means programming with asynchronous data streams: you declare how data flows and transforms, and you react to events as they arrive (Observer pattern + iterator + functional operators).
RxJava is the JVM implementation of ReactiveX. Core parts: Observable (source) → Operators (transform) → Observer/Subscriber (consumer), with Schedulers for threading and Disposables for cleanup. RxAndroid adds AndroidSchedulers.mainThread().
Answer
| Type | Emits | Use case | Coroutine equivalent |
|---|---|---|---|
Observable |
0..N items, no backpressure | UI events, small streams | Flow |
Flowable |
0..N items, with backpressure | Large/fast streams (DB, sensors) | Flow |
Single |
exactly 1 item or error | Network call | suspend fun(): T |
Maybe |
0 or 1 item, or error | Cache lookup | suspend fun(): T? |
Completable |
no item; completes or errors | Save/delete operation | suspend fun() |
Answer
onSubscribe(Disposable) → onNext(item)* → then either onComplete() or onError(Throwable). Nothing comes after a terminal event. Single uses onSuccess and onError.
Answer
subscribeOn(Schedulers.io())– chooses the thread where the source starts (where the subscription happens). Its position in the chain doesn't matter, and only the first one counts.observeOn(AndroidSchedulers.mainThread())– switches the thread for every operator downstream of it. You can use it multiple times.
api.getUsers()
.subscribeOn(Schedulers.io())
.map { it.toDomain() }
.observeOn(AndroidSchedulers.mainThread())
.subscribe({ render(it) }, { showError(it) })Answer
Schedulers.io()– unbounded cached pool for I/O.Schedulers.computation()– fixed pool (one thread per CPU core) for CPU work.Schedulers.newThread(),Schedulers.single(),Schedulers.trampoline()(current thread, queued; useful in tests),Schedulers.from(executor).AndroidSchedulers.mainThread()(RxAndroid).
Answer
map– transforms each item 1:1, synchronously.flatMap– maps each item to an Observable and merges the results. Runs concurrently, order not guaranteed.concatMap– like flatMap but sequential: waits for each inner Observable, so order is preserved.switchMap– cancels the previous inner Observable when a new item arrives. Ideal for search-as-you-type.
Answer
zip– pairs items by index from multiple sources (wait for both API calls, then combine).merge– interleaves emissions from sources as they arrive.concat– subscribes to the sources one after another (cache first, then network).combineLatest– emits whenever any source emits, using the latest from each (form validation).
Answer
A Subject is both an Observable and an Observer (hot). It's used to bridge imperative code into Rx.
- PublishSubject – emits only items received after subscription.
- BehaviorSubject – emits the latest item (or a default) on subscribe, then the following ones. Like
StateFlow. - ReplaySubject – replays all (or N) previous items to new subscribers.
- AsyncSubject – emits only the last value, on completion.
Relays (RxRelay) are Subjects that can't terminate, which is safer for UI events.
Answer
- Cold – starts producing per subscriber (each subscription re-runs the source, e.g. a Retrofit call).
- Hot – produces regardless of subscribers, and they share the emissions (Subjects, UI clicks).
- Turn cold into hot with
publish().refCount(),share(),replay(1).autoConnect().
Answer
Backpressure is the situation where the producer emits faster than the consumer can handle. An Observable has no backpressure support, so it risks OutOfMemoryError. A Flowable uses request-based flow control with strategies: BackpressureStrategy.BUFFER, DROP, LATEST, ERROR, MISSING, or operators like onBackpressureDrop().
Answer
subscribe() returns a Disposable. Calling dispose() unsubscribes and stops the work. If you don't dispose, the subscription keeps the Activity alive, which causes memory leaks and callbacks on dead views.
private val disposables = CompositeDisposable()
disposables += api.getUsers().subscribe(...)
override fun onCleared() { disposables.clear() } // in ViewModelclear() disposes everything and lets you reuse the container; dispose() also makes it reject future additions.
Answer
onErrorReturn { default },onErrorReturnItem(x).onErrorResumeNext { fallbackObservable }.retry(3),retryWhen { errors -> ... }(backoff).doOnError { log(it) }for side effects only.- Always provide an
onErrorinsubscribe. Otherwise you get anOnErrorNotImplementedExceptioncrash. - Set
RxJavaPlugins.setErrorHandlerfor undeliverable exceptions (errors after disposal).
Answer
debounce(300ms)– emits only after a quiet period (search input).throttleFirst(1s)– emits the first item, then ignores the rest for a window (prevents double clicks).distinctUntilChanged()– skips consecutive duplicates.
Answer
Side-effect operators that don't change the stream: logging, showing or hiding a loader (doOnSubscribe { showLoading() }, doFinally { hideLoading() }).
Answer
TestObserver:observable.test().assertValues(a, b).assertComplete().- Replace schedulers:
RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }andRxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }(usually in a JUnit rule). TestSchedulerfor controlling time (advanceTimeBy).
Answer
| RxJava | Coroutines |
|---|---|
Single<T> |
suspend fun(): T |
Completable |
suspend fun() |
Observable/Flowable |
Flow<T> |
BehaviorSubject |
MutableStateFlow |
PublishSubject |
MutableSharedFlow |
subscribeOn/observeOn |
flowOn / withContext |
CompositeDisposable |
Structured concurrency (scope cancellation) |
switchMap |
flatMapLatest |
Migrate incrementally, layer by layer with kotlinx-coroutines-rx3 (.await(), .asFlow(), rxSingle {}, asObservable()). Start with the data layer or new features.
Answer
just(x)– emits a value that's already computed (evaluated eagerly when the Observable is built, even if nobody subscribes).Observable.just(db.query())runs the query immediately, possibly on the main thread.fromCallable { }– computes the value lazily on subscribe, on thesubscribeOnthread. Exceptions becomeonError. The best choice for wrapping a single blocking call.create { emitter -> }– full manual control: emit many values, complete or fail, wrap listener/callback APIs, and register cleanup withemitter.setCancellable { }. You must respect the Observable contract yourself.defer { Observable.just(x) }– creates a fresh Observable for each subscriber at subscribe time, so each subscriber gets current data. It's useful to make an eager source lazy.
Answer
Observable.timer(2, SECONDS)– emits a single0Lafter the delay, then completes (one-shot delayed actions, splash timeouts)..delay(2, SECONDS)– an operator that shifts each emission of an existing stream by the given time. (Errors aren't delayed by default.)Observable.interval(1, SECONDS)– emits0, 1, 2…repeatedly at a fixed period (polling, countdowns; combine withtake(n)). Dispose it, or it runs forever.
All of them run on Schedulers.computation() by default. Coroutine equivalents: delay(), and loops with delay inside flow {}.
Answer
// Parallel calls → combine
Single.zip(api.getProfile().subscribeOn(Schedulers.io()),
api.getFeed().subscribeOn(Schedulers.io())) { profile, feed -> Home(profile, feed) }
.observeOn(AndroidSchedulers.mainThread())
.subscribe(::render, ::showError)
.addTo(disposables)
// Instant search
searchSubject
.debounce(300, TimeUnit.MILLISECONDS)
.map { it.trim() }
.filter { it.isNotEmpty() }
.distinctUntilChanged()
.switchMapSingle { q -> api.search(q).subscribeOn(Schedulers.io()).onErrorReturnItem(emptyList()) }
.observeOn(AndroidSchedulers.mainThread())
.subscribe(::showResults)Each source needs its own subscribeOn(io()); otherwise zip subscribes to them sequentially on the same thread. switchMap cancels stale searches. Errors are handled inside the inner stream so the search stream survives.
⬅ Previous: Threading & Background Work · ⬅ Back to index · Next: Performance & Memory ➡