How perform long calculations on background thread in RXJava Android

android, java, multithreading, rx-java

Solution

you will not achieve concurrency using map-Operator.

The first subscribeOn will move all emitions to IO-scheduler. No concurrency happening here.

.subscribeOn(Schedulers.io())

The map-operator will be called synchronously from previous thread. In your case it would be some thread from IO-threadpool.

.map(new Func1<String, String>() {

After the map-Operator has been executed, you will move the value from the IO-thread to the Android-UI-event-loop with

.observeOn(AndroidSchedulers.mainThread())

After the value has been translated from IO-thread to the UI-thread the next value from the initial observable will be processed.

Observable.just("true")

In your example there will be no more values because you only produce one value.

In order to achieve concurrency you should use flatMap instead of map. And use subscribeOn() in the flatMap to create each stream on another thread.

Please consider this example, to see how concurrency is happening. Every observable will be subscribed at once, so the max. time for the teset would be something around 5 seconds. If now concurrency would happen it would take 1+2+3+4+5 seconds plus execution time.

@Test
public void name1() throws Exception {

        Observable<Integer> value = Observable.just(1_000, 2_000, 3_000, 4_000, 5_000)
                .flatMap(i -> Observable.fromCallable(() -> doWork(i)).subscribeOn(Schedulers.io())
                ).doOnNext(integer -> System.out.println("value"));

        value.test().awaitTerminalEvent();
}

private int doWork(int sleepMilli) {
        try {
            Thread.sleep(sleepMilli);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return -1;
}

If you want to know more about how concurrency is happening with flatMap please consider reading http://tomstechnicalblog.blogspot.de/2015/11/rxjava-achieving-parallelization.html

In regard to your code I would suggest:

- Move the anonymouse implementation of interface to a private inner class implementation and use an instance of it. You will get a more readable observable

Don't use side-effects to global variables from operators within. You will get race-condition if concurrency is involved.

List<FeedsListModel> eventFeedItems = Arrays.asList(new FeedsListModel(), new FeedsListModel());

Observable<FeedsListModel> feedsListModelObservable = Observable.fromIterable(eventFeedItems)
                    .flatMap(feedsListModel -> Observable.fromCallable(() -> calculation(feedsListModel))
                            .subscribeOn(Schedulers.computation())
);

feedsListModelObservable
        .toList()
        .observeOn(Schedulers.io())
        .subscribe(feedsListModels -> {
            // do UI stuff
});

Helping-method:

private FeedsListModel calculation(FeedsListModel model) {
    // do calculation here

    return new FeedsListModel();
}

Problem

Original source