Android RxJava3 核心使用场景与实战指南

关键点解析：

• subscribeOn(Schedulers.io())：指定上游线程为 IO 线程，适合网络请求。
• observeOn(AndroidSchedulers.mainThread())：指定下游线程为主线程，适合 UI 更新。
• flatMap：将上游发射的数据映射为新的 Observable，实现逻辑上的'下一步'。

2. 多任务并行合并

场景描述：当两个接口相互独立，但业务逻辑要求必须等待两个接口都返回数据后才能进行后续处理（例如同时加载用户信息和配置信息）。

实现方案：使用 zip 操作符。zip 会将两个 Observable 发射的数据按顺序配对，直到其中一个完成。如果只需要任意一个完成即可，可使用 merge；如果需要严格配对，使用 zip。

private void zipRequest() {
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
    Observable<BaseResponse<List<HotKeyResp.DataBean>>> hotKey = ApiManager.getInstance().getApiService().getHotKey();

    Observable.zip(articleList, hotKey, (resp1, resp2) -> {
        // 合并规则：接收两个参数，返回组合后的结果
        if (resp1 != null && resp2 != null) {
            return resp1.toString() + resp2.toString();
        }
        return "";
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {}

        @Override
        public void onNext(@NonNull String msg) {
            if (!TextUtils.isEmpty(msg)) {
                mTv.setText(msg);
            }
        }

        @Override
        public void onError(@NonNull Throwable e) {}

        @Override
        public void onComplete() {}
    });
}

关键点解析：

• zip 操作符要求两个 Observable 发射的数据数量一致，否则会提前终止。
• 返回值通过 Lambda 表达式定义，便于处理多源数据。

二、轮询与定时任务

轮询机制常用于实时通知、状态检查等场景。RxJava3 提供了多种操作符来实现灵活的轮询策略。

1. 固定次数轮询（依赖上一次结果）

场景描述：每隔一定时间发起一次请求，直到达到指定次数或收到特定成功标志。每次请求需等待上一次完成。

实现方案：结合 repeatWhen 和 takeUntil。repeatWhen 允许在错误发生时决定是否需要重试，而 takeUntil 用于设定终止条件。

private void timedPolling(int pollingTimes) {
    AtomicInteger times = new AtomicInteger();
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();

    articleList.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(Observable<Object> objectObservable) throws Throwable {
            // 将重复信号转换为延迟发射的信号
            return objectObservable.flatMap(o -> 
                Observable.just(0).delay(2, TimeUnit.SECONDS)
            );
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    // 终止条件：达到次数 或 接口返回成功码
    .takeUntil(response -> times.incrementAndGet() >= pollingTimes || response.getErrorCode() == 0)
    .subscribe(new Observer<BaseResponse<ArticleListResp>>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {}

        @Override
        public void onNext(@NonNull BaseResponse<ArticleListResp> response) {
            // 处理轮询结果
        }

        @Override
        public void onError(@NonNull Throwable e) {}

        @Override
        public void onComplete() {}
    });
}

2. 不依赖结果的间隔轮询

场景描述：无论上一次请求是否成功，都按照固定频率发起新请求。常用于心跳检测或实时数据拉取。

实现方案：使用 intervalRange 或 interval 生成时间序列，再通过 flatMap 转换为网络请求。

private void intervalPolling(int pollTimes) {
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
    
    Observable.intervalRange(0, pollTimes, 0, 2000, TimeUnit.MILLISECONDS)
            .flatMap(aLong -> articleList)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
                @Override
                public void accept(BaseResponse<ArticleListResp> response) throws Throwable {
                    // 处理结果
                    // 如需提前停止，可调用 mDisposable.dispose()
                }
            });
}

3. 无限轮询与动态取消

场景描述：持续轮询直到满足特定业务条件（如接口返回 code=0）。

实现方案：使用 takeUntil 监听响应内容，一旦满足条件即自动取消订阅。

Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
Disposable mDisposable = Observable.interval(0, 2000, TimeUnit.MILLISECONDS)
        .flatMap(aLong -> articleList)
        .takeUntil(response -> response.getErrorCode() == 0)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
            @Override
            public void accept(BaseResponse<ArticleListResp> response) throws Throwable {
                // 处理回调
            }
        });

注意事项：务必在 Activity 或 Fragment 销毁时（如 onDestroy）调用 mDisposable.dispose()，防止内存泄漏。

三、UI 交互效果实现

RxJava3 结合 RxBinding 可以极大地简化 UI 事件的处理，实现倒计时、打字机效果等常见交互。

1. 验证码倒计时

利用 intervalRange 生成时间序列，配合 map 转换显示文本。

private void countDown(int countDownSeconds) {
    Observable.intervalRange(0, countDownSeconds, 0, 1000, TimeUnit.MILLISECONDS)
            .map(aLong -> (countDownSeconds - aLong) + "s 后重新获取")
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    mTv.setEnabled(false);
                }

                @Override
                public void onNext(@NonNull String s) {
                    mTv.setText(s);
                }

                @Override
                public void onError(@NonNull Throwable e) {
                    mTv.setEnabled(true);
                    mTv.setText("获取验证码");
                }

                @Override
                public void onComplete() {
                    mTv.setText("获取验证码");
                    mTv.setEnabled(true);
                }
            });
}

2. 打字机效果

通过控制文本长度逐步绘制，模拟逐字输入的效果。

@RequiresApi(api = Build.VERSION_CODES.M)
public class DaziView extends View {
    private TextPaint mTextPaint;
    private StaticLayout mStaticLayout;

    public DaziView(Context context) {
        super(context);
        initTextPaint();
    }

    private void initTextPaint() {
        mTextPaint = new TextPaint(Paint.ANTI_ALIAS_FLAG);
        mTextPaint.setTextSize(48);
        mTextPaint.setColor(Color.parseColor("#000000"));
    }

    public void drawText(String content) {
        if (!TextUtils.isEmpty(content)) {
            Observable.intervalRange(0, content.length() + 1, 0, 150, TimeUnit.MILLISECONDS)
                    .subscribe(new Consumer<Long>() {
                        @Override
                        public void accept(Long aLong) throws Throwable {
                            // 动态改变文本长度
                            mStaticLayout = StaticLayout.Builder.obtain(
                                    content, 0, aLong.intValue(), mTextPaint, getWidth())
                                    .build();
                            invalidate();
                        }
                    });
        }
    }

    @Override
    protected void onDraw(Canvas canvas) {
        super.onDraw(canvas);
        if (mStaticLayout != null) {
            mStaticLayout.draw(canvas);
        }
    }
}

四、结合 RxBinding 的使用场景

RxBinding 将 Android View 的事件转换为 Observable，使得 UI 事件处理符合响应式范式。

1. 点击事件防抖

防止用户快速连续点击导致重复提交。

RxView.clicks(button)
        .throttleFirst(1000, TimeUnit.MILLISECONDS)
        .subscribe(unit -> {
            // 一秒内第一次点击有效
            handleButtonClick();
        });

2. 搜索框输入优化

对输入流进行节流（debounce）和跳过初始空值处理。

RxTextView.textChanges(editText)
        .debounce(1000, TimeUnit.MILLISECONDS) // 1 秒无新输入才发射
        .skip(1)                               // 跳过初始空内容
        .subscribeOn(AndroidSchedulers.mainThread())
        .subscribe(charSequence -> {
            // 执行搜索逻辑
            performSearch(charSequence.toString());
        });

3. 联合判断

监听多个输入框的状态变化，只有当所有条件满足时才启用按钮。

Observable<CharSequence> observableEdittext = RxTextView.textChanges(editText).skip(1);
Observable<CharSequence> observableEdittextTwo = RxTextView.textChanges(editText_two).skip(1);

Observable.combineLatest(observableEdittext, observableEdittextTwo, (str1, str2) -> {
    return !TextUtils.isEmpty(str1) && !TextUtils.isEmpty(str2);
}).subscribe(enabled -> {
    button.setEnabled(enabled);
});

五、内存泄漏防护

RxJava 订阅若未正确管理生命周期，极易导致 Activity 或 Fragment 无法回收，引发内存泄漏。

1. 使用 CompositeDisposable 统一管理

推荐将所有订阅关系放入 CompositeDisposable，并在销毁时统一清除。

private CompositeDisposable compositeDisposable = new CompositeDisposable();

// 订阅时添加
compositeDisposable.add(RxView.clicks(button)
        .subscribe(unit -> handleButtonClick()));

// 销毁时清除
@Override
protected void onDestroy() {
    super.onDestroy();
    if (!compositeDisposable.isDisposed()) {
        compositeDisposable.clear();
    }
}

2. 避免在后台线程持有 Context 引用

确保在 observeOn 切换回主线程后再访问 UI 控件，且不要在闭包中强引用 Activity。

3. 及时取消订阅

对于长生命周期的 Observable（如轮询），务必在业务结束或组件销毁时主动调用 dispose()。

六、总结

RxJava3 为 Android 异步编程提供了强大的工具集。通过合理使用 flatMap、zip、repeatWhen 等操作符，开发者可以构建出逻辑清晰、易于维护的代码结构。同时，结合 RxBinding 和生命周期管理最佳实践，可以有效提升应用性能和稳定性。在实际项目中，建议遵循以下原则：

1. 明确线程调度：IO 线程处理耗时操作，主线程更新 UI。
2. 规范错误处理：每个订阅都应包含 onError 处理逻辑。
3. 严格生命周期管理：使用 CompositeDisposable 或 AutoDispose 等库管理订阅。
4. 避免过度封装：保持代码可读性，不要为了使用 RxJava 而强行引入复杂链式调用。

掌握这些核心场景与技巧，将有助于开发者更高效地应对复杂的业务需求，打造高质量的 Android 应用。