操作线程RxJava的有效方法

Ant*_*nko 5 java multithreading android rx-java

在我的项目中,我需要处理不同线程中的对象.为了操纵流的行为,我创建了新的observable来改变它们的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
                });
Run Code Online (Sandbox Code Playgroud)

但我认为在RxJava中有更多美观有效的方法来处理不同线程中的一个响应.我试着谷歌,但我没有找到任何东西.

谢谢,
安东

Glu*_*uck 3

在 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
            });
Run Code Online (Sandbox Code Playgroud)

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

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

请注意,AFAICT 您的代码在没有该行的map(Observable::just)情况下也可以正常工作,因为“observeOn”语句只会影响进一步向下的流(而不是之前的“do”语句)