RxJava 操作线程的高效方式

Question

在我的项目中，我需要在不同的线程中处理对象。为了操纵流的行为，我创建了新的 observables 以这种方式改变它们的 observeOn()：

apiService.getObjects(token) // Retrofit
                .compose(bindToLifecycle())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(o -> {
                    // process in Main Thread
                })
                .map(Observable::just)  // create new one, to change thread for it
                .observeOn(Schedulers.io()) 
                .subscribe(o -> {
                    // process in the background thread
                });

但我认为在 RxJava 中有更漂亮和有效的方式来处理不同线程中的一个响应。我尝试 google 它，但我没有找到任何东西。

谢谢，
安东

Answer 1

在 Rx 中，通常建议避免 'do' 块中的副作用（只有在流被订阅时才会执行），并且更喜欢订阅代码。

在您的情况下，您可以利用 cache() 或 publish()...connect()，例如：

query = apiService.getObjects(token)
            .compose(bindToLifecycle())
            .subscribeOn(Schedulers.io())
            .cache();

query.observeOn(AndroidSchedulers.mainThread())
            .subscribe(o -> {
                // process in Main Thread
            })
query.observeOn(Schedulers.io())
            .subscribe(o -> {
                // process in the background thread
            });

使用 publish() 而不是 cache()，代码是相同的，但您可以通过连接流来决定何时触发查询（您调用 query.connect() 连接 2 个订阅后）。

如果您的订阅工作是后台计算，Schedulers.computation() 可能优于 Schedulers.io()。

请注意，AFAICT 您的代码在没有 map(Observable::just) 行的情况下也能正常工作，因为 'observeOn' 语句只会影响更下方的流（而不是之前的 'do' 语句）

RxJava 操作线程的高效方式

Efficient way to manipulate threads RxJava

java

multithreading

android

rx-java