RxJava(Android)中PublishProcessor无法正常工作的问题咨询

嘿，我来帮你搞定这个需求！你要的功能核心就是用PublishProcessor做事件入口，滚动时丢进去Integer初始数据，每个数据启动后台下载，完了返回String结果，还得一直能接新数据对吧？

先说说你可能踩的坑 #

你说测试代码跑不起来，大概率是这几个原因：

• 没搞对线程调度：后台下载没放到IO线程，或者结果没切回合适的线程（比如Android主线程）
• 订阅关系没保住：比如Disposable没持有，导致订阅被回收了
• 处理器意外挂了：比如不小心调用了onComplete()，导致它不再接收新事件
• 没处理异常：单个下载任务报错就把整个处理器搞挂了，后续事件没法处理

完整可运行的示例代码 #

下面是符合你需求的完整代码，每一步我都给你标清楚为啥这么写：

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.processors.PublishProcessor;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class ScrollDownloadHandler {
    // 核心：创建PublishProcessor作为滚动事件的入口，接收Integer类型的初始数据
    private final PublishProcessor<Integer> scrollEventProcessor = PublishProcessor.create();

    public ScrollDownloadHandler() {
        // 配置处理器的处理逻辑，这是核心
        scrollEventProcessor
                // 用flatMap而不是map！因为每个初始数据要启动独立的后台任务，flatMap能并发处理，不阻塞新事件
                .flatMap(initialData -> {
                    // 把初始数据丢去做后台下载
                    return simulateDownload(initialData)
                            // 下载任务放IO线程，别占主线程
                            .subscribeOn(Schedulers.io())
                            // 单个任务失败了别搞垮整个处理器，返回个错误提示就行
                            .onErrorReturn(error -> "数据" + initialData + "下载失败：" + error.getMessage());
                })
                // 结果回调到你需要的线程，Android的话换成AndroidSchedulers.mainThread()
                .observeOn(Schedulers.single())
                // 订阅上，处理结果和全局异常
                .subscribe(
                        downloadResult -> System.out.println("搞定！" + downloadResult),
                        globalError -> System.err.println("处理器出大问题了：" + globalError.getMessage())
                );
    }

    // 模拟真实的后台下载任务，你把这里换成你的网络请求就行
    private Flowable<String> simulateDownload(Integer initialData) {
        return Flowable.fromCallable(() -> {
            // 模拟下载耗时，比如1秒
            Thread.sleep(1000);
            return "针对初始数据" + initialData + "的下载内容";
        });
    }

    // 对外暴露的方法，用户滚动时就调用这个传初始数据
    public void onUserScroll(Integer initialData) {
        scrollEventProcessor.onNext(initialData);
    }

    // 测试用的main方法，跑起来就能看到效果
    public static void main(String[] args) throws InterruptedException {
        ScrollDownloadHandler handler = new ScrollDownloadHandler();

        // 模拟用户连续滚动，传3个初始数据
        handler.onUserScroll(1);
        handler.onUserScroll(2);
        handler.onUserScroll(3);

        // 等所有下载完成
        Thread.sleep(3000);
    }
}

关键细节得注意 #

• 为啥用flatMap？：如果用map的话，后台任务会在订阅线程跑，容易阻塞，新的滚动事件就接不住了。flatMap能把每个Integer转成独立的Flowable，并发执行，完全不影响后续事件的接收。
• 线程调度不能少：subscribeOn指定下载任务的线程，observeOn指定结果回调的线程，这俩是异步处理的关键。
• 异常要兜住：onErrorReturn保证单个下载失败不会导致整个PublishProcessor终止，后面的滚动事件照样能处理。
• 订阅要管好：实际项目里（比如Android），要把订阅返回的Disposable存起来，页面销毁时调用dispose()，防止内存泄漏。

跑起来的效果 #

运行main方法，你会看到类似输出（因为是并发，顺序可能有点不一样）：

搞定！针对初始数据1的下载内容
搞定！针对初始数据2的下载内容
搞定！针对初始数据3的下载内容

这样就完全满足你的需求了：滚动时传初始数据，后台异步下载，返回结果，还能一直接收新的初始数据~

内容的提问来源于stack exchange，提问作者bagins82