RxJava Questions About Threads, CompositeDisposable and Scheduler in an Android Application

Question

My code is like this:

timingObservable = getCurrentModule()
                .zipWith(Observable.interval(200, TimeUnit.MILLISECONDS), (currentModule, interval) -> currentModule)
                .repeat()
                .distinctUntilChanged()
                .getModuleDataFromDb()

compositeDisposable.add(timingObservable
        .subscribeOn(Schedulers.io())
        .subscribe(next -> {
            .
            .
            .
        }));

public Observable<String> getCurrentModule() {
    return Observable.fromCallable(() -> {
        String currentModule = "";
        // doing some none database work and computation
        .
        .
        .
        return currentModule;
    }
}

It is supposed to check the current module periodically and get some data from db if the module is changed. I have several questions:

• In the RxThreadFactory class of RxJava and in the newThread() method we have the line t.setDaemon(true), so is it true that all RxJava threads are daemon threads? So, they are alive as long as a component of app is alive and the app process is still running, right?

• I am adding the disposable return of subscribe() to a compositeDisposable and call dispose in onDestory() of my Service/Activity classes. What happens to those disposables when the service gets killed without onDestroy() being called. I mean, since the compositeDisposable object is destoryed, is it possible that I lose the ability to dispose disposables? Should I hold application wide instance of CompositeDisposable?

• In terms of performance, which one is recommended in this code? subscribeOn(Schedulers.io()) or subscribeOn(Schedulers.computation()) since Observable.interval uses computation scheduler by default and we have DB work too.

• Any suggestions to improve the above code for periodic tasks?

Answer 1

To answer the questions :

In the RxThreadFactory class of RxJava and in the newThread() method we have the line t.setDaemon(true), so is it true that all RxJava threads are daemon threads? So, they are alive as long as a component of app is alive and the app process is still running, right?

In Java Thread::setDaemon(true) simply means that once all non-daemon threads have finished then these "daemon" threads are abandoned and the JVM shut down. For reference the android "main" Thread is not a daemon thread, but has a Looper. Daemon threads can naturally finish and do not stop the process from exiting. You should not rely on this mechanism for long running tasks and use foreground services and/or WorkManager, Rx thread pools could last as long as the process they are running in, unless the Executor they are tied to is explicitly shutdown.

I am adding the disposable return of subscribe() to a compositeDisposable and call dispose in onDestory() of my Service/Activity classes. Lets say in a scenario there are one service and one activity and compositeDisposable belongs to the service. What happens to those disposables when the service gets killed without onDestroy() being called and activity remains alive. I mean, since the compositeDisposable object is destoryed, is it possible that I lose the ability to dispose disposables? Should I hold application wide instance of CompositeDisposable?

A Service would only get destroyed without the lifecycle call backs if 1) Android kills the process to reclaim resources, in that case its not relevant to clear resources, or 2) The program crashes, and again resources cleanup is not required

In terms of performance, which one is recommended in this code? subscribeOn(Schedulers.io()) or subscribeOn(Schedulers.computation()) since Observable.interval uses computation scheduler by default and we have DB work too.

Schedulers.io is an unbounded thread pool, whilst Schedulers.computation() is bounded (a pool of 8 threads I believe). Performance could vary, but in most cases negligible difference. One scenario Schedulers.computation() may be slower could be if you have a lot of concurrency using this thread pool continuously, meaning you are waiting for a Thread to become free. In contrast Schedulers.io() could have a upfront cost of creating new threads from the ThreadFactory it uses. However it will try to use existing thread from its pool. Metrics would be needed to really see any performance differences for individual use cases. By general rule io() suggests that for work such as file/database/networking should be done using this Scheduler and computation work like timers/algorithms should use the latter.

Any suggestions to improve the above code for periodic tasks?

As already suggested polling, generally is not a good idea, and Reactive Streams by concept is more useful as a pub/sub observer pattern. However it seems you have some restrictions as to what you can modify. With the provided code there is far too little context to really give any concrete improvements, and only suggestions, which will ultimately end up with "I can't do that because...".