RxJava3如何让onNext内任务运行在后台线程避免UI卡顿

问题根因 #

• 核心错误是调用了.observeOn(AndroidSchedulers.mainThread())，这个操作符会将后续所有下游操作（包括onNext/onError/onComplete回调、以及写在该操作符之后的所有变换类操作符如filter/map等）的执行线程指定为主线程，所以在此之后无论是在onNext里写耗时逻辑，还是添加filter/sleep逻辑，都会运行在主线程导致卡顿。
• 你提到把Thread.sleep(5000)放在filter里依然卡顿，大概率是你将filter操作符写在了observeOn(AndroidSchedulers.mainThread())之后，自然会执行在主线程。

解决方案 #

场景1：耗时计算完成后才需要更新UI #

把所有耗时逻辑放在observeOn(mainThread)之前，指定在后台线程执行，处理完结果后再切主线程做UI操作即可，示例代码：

private void start(){
    reactiveDataSource.getObservableList()
        .subscribeOn(Schedulers.io())
        // 所有耗时操作都放在observeOn(mainThread)之前，执行在IO线程
        .filter(item -> {
            // 这里的sleep/计算逻辑都跑在IO线程，不会卡主线程
            Thread.sleep(5000);
            return !item.isEmpty();
        })
        .map(item -> {
            // 这里放你的密集计算逻辑，执行在IO线程
            String processedResult = "处理后的" + item;
            return processedResult;
        })
        // 所有计算完成后，再切主线程更新UI
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "subscribed in observable list");
            }

            @Override
            public void onNext(String processedItem) {
                // 这里只有UI更新逻辑，无耗时操作
                Log.e(TAG, "in observable list ->" + processedItem);
                if(processedItem.contains("D")){
                    Toast.makeText(getApplicationContext(), Thread.currentThread().toString(), Toast.LENGTH_SHORT).show();
                } else if(processedItem.contains("E")){
                    Toast.makeText(getApplicationContext(), "E", Toast.LENGTH_SHORT).show();
                } else if(processedItem.contains("F")){
                    Toast.makeText(getApplicationContext(), "F", Toast.LENGTH_SHORT).show();
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "error in observable list " + e.getMessage());
                Toast.makeText(getApplicationContext(), e.toString(), Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "in observable list complete");
                Toast.makeText(getApplicationContext(), "Complete!!!", Toast.LENGTH_SHORT).show();
            }
        });
}

场景2：onNext里同时有耗时逻辑和UI操作，不需要提前处理 #

不要全局切到主线程，在需要更新UI的地方单独切主线程即可，示例代码：

private void start(){
    reactiveDataSource.getObservableList()
        .subscribeOn(Schedulers.io())
        // 不切主线程，所有回调默认执行在IO线程
        .subscribe(new Observer<String>() {
            private final Handler mainHandler = new Handler(Looper.getMainLooper());
            
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "subscribed in observable list");
            }

            @Override
            public void onNext(String item) {
                // 这里执行耗时计算，跑在IO线程，不会卡主线程
                Log.e(TAG, "计算线程：" + Thread.currentThread().toString());
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                String processed = "处理后:" + item;
                
                // 需要更新UI的时候，手动切到主线程
                mainHandler.post(() -> {
                    Toast.makeText(getApplicationContext(), processed, Toast.LENGTH_SHORT).show();
                });
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "error in observable list " + e.getMessage());
                mainHandler.post(() -> 
                    Toast.makeText(getApplicationContext(), e.toString(), Toast.LENGTH_SHORT).show()
                );
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "in observable list complete");
                mainHandler.post(() -> 
                    Toast.makeText(getApplicationContext(), "Complete!!!", Toast.LENGTH_SHORT).show()
                );
            }
        });
}

注意事项 #

如果是CPU密集型的计算任务，建议用Schedulers.computation()而不是Schedulers.io()，前者是专门为计算任务优化的线程池。

内容的提问来源于stack exchange，提问作者LIGHTNING RED DRAGON GAMING