文章目录1.1 与 Retrofit 结合1.2 与 MVP/MVVM 架构结合1.2.1 与 MVP (Model-View-Presenter) 结合1.2.2 RxJava 与 MVVM (Model-View-ViewModel) 结合1.3. RxBus (事件总线)1.4. 错误处理与资源管理1.5 复杂任务编排1.5.1 并行任务合并zip 操作符1.5.2. 顺序任务合并concat 操作符1.5.3. 动态数据流合并merge 操作符1.5.4. 最新数据合并combineLatest 操作符1.6 UI 事件处理1.6.1 防抖优化搜索输入将 RxJava 融入到实际项目架构中解决复杂问题。1.1 与 Retrofit 结合Retrofit 官方支持返回 Observable 或 Flowable是处理网络请求的黄金搭档。public interface ApiService { GET(users/{id}) ObservableUser getUser(Path(id) int id); } // 使用 apiService.getUser(123) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( user - { /* 处理成功 */ }, error - { /* 处理错误 */} );1.2 与 MVP/MVVM 架构结合将 RxJava 与 MVP 或 MVVM 架构结合使用是现代 Android以及 Java Swing开发中的最佳实践之一。这种结合能充分发挥各自的优势架构模式负责清晰的职责分离和可测试性而 RxJava 则优雅地处理异步、事件流和复杂的线程切换。1.2.1 与 MVP (Model-View-Presenter) 结合MVP 的核心思想是将 Activity/Fragment (View) 从繁重的业务逻辑中解放出来使其只负责 UI 的展示和用户交互的传递。Presenter 作为中间层负责处理业务逻辑并与 Model 层交互。// 1. 定义 View 接口publicinterfaceUserView{voidshowLoading();voidhideLoading();voidshowUsers(ListUserusers);voidshowError(Stringmessage);}// 2. Model 层 (返回 RxJava 数据流)publicclassUserModel{privateApiServiceapiService;// 假设使用 RetrofitpublicObservableListUsergetUsers(){returnapiService.getUsers()// 返回 ObservableListUser.subscribeOn(Schedulers.io());// 在 IO 线程执行网络请求}}// 3. Presenter 层 (核心: 处理 RxJava 链)publicclassUserPresenter{privateUserViewview;privateUserModelmodel;privateCompositeDisposabledisposablesnewCompositeDisposable();publicUserPresenter(UserViewview,UserModelmodel){this.viewview;this.modelmodel;}publicvoidloadUsers(){view.showLoading();// 建立 RxJava 订阅disposables.add(model.getUsers().observeOn(AndroidSchedulers.mainThread())// 切换到主线程更新 UI.subscribe(users-{view.hideLoading();view.showUsers(users);// 成功: 更新 UI},error-{view.hideLoading();view.showError(加载失败: error.getMessage());// 失败: 显示错误}));}// 在 Activity/Fragment onDestroy 时调用防止内存泄漏publicvoidonDestroy(){disposables.clear();}}// 4. View 层 (Activity/Fragment)publicclassUserActivityextendsAppCompatActivityimplementsUserView{privateUserPresenterpresenter;OverrideprotectedvoidonCreate(BundlesavedInstanceState){super.onCreate(savedInstanceState);setContentView(R.layout.activity_user);// 初始化 PresenterpresenternewUserPresenter(this,newUserModel());// 触发加载presenter.loadUsers();}OverrideprotectedvoidonDestroy(){super.onDestroy();presenter.onDestroy();// 关键: 取消所有订阅}// 实现 View 接口方法OverridepublicvoidshowLoading(){// 显示进度条}OverridepublicvoidhideLoading(){// 隐藏进度条}OverridepublicvoidshowUsers(ListUserusers){// 更新 RecyclerView 或 ListView}OverridepublicvoidshowError(Stringmessage){// 弹出 Toast 或 Snackbar}}优势:职责清晰: View 只管展示Presenter 处理逻辑和数据流。易于测试: Presenter 不依赖 Android 组件可以方便地进行单元测试。异步处理优雅: RxJava 完美解决了网络请求、数据库操作等异步问题。挑战:内存泄漏: 必须妥善管理 Disposable在生命周期结束时取消订阅。Presenter 膨胀: 如果业务逻辑复杂Presenter 可能会变得非常庞大。可以通过引入 Interactor (或称 Use Case) 层来分担业务逻辑。1.2.2 RxJava 与 MVVM (Model-View-ViewModel) 结合MVVM 通过 数据绑定 (Data Binding) 或 LiveData/StateFlow 将 View 与 ViewModel 解耦。View 通过观察 ViewModel 中的数据变化来自动更新 UIViewModel 则负责准备和管理这些数据。RxJava 可以作为 ViewModel 内部处理异步数据流的强大工具最终将结果暴露给 View。由于 Android 官方推荐在 MVVM 中使用 LiveData而 LiveData 本身不是响应式流我们通常使用 LiveDataReactiveStreams 工具类进行桥接。// 1. Model 层 (同 MVP)publicclassUserModel{publicObservableListUsergetUsers(){returnapiService.getUsers().subscribeOn(Schedulers.io());}}// 2. ViewModel 层 (核心: 使用 RxJava 处理逻辑输出 LiveData)publicclassUserViewModelextendsViewModel{privateUserModelmodel;// 暴露给 View 的 LiveDataprivateMutableLiveDataListUserusersLiveDatanewMutableLiveData();privateMutableLiveDataBooleanloadingLiveDatanewMutableLiveData();privateMutableLiveDataStringerrorLiveDatanewMutableLiveData();// 提供 LiveData 给 View 观察publicLiveDataListUsergetUsers(){returnusersLiveData;}publicLiveDataBooleanisLoading(){returnloadingLiveData;}publicLiveDataStringgetError(){returnerrorLiveData;}publicUserViewModel(UserModelmodel){this.modelmodel;}publicvoidloadUsers(){loadingLiveData.setValue(true);// 将 RxJava Observable 转换为 LiveDataLiveDataListUserliveDataLiveDataReactiveStreams.fromPublisher(model.getUsers().toFlowable(BackpressureStrategy.LATEST)// 转换为 Flowable 以支持背压.observeOn(AndroidSchedulers.mainThread())// 确保在主线程发射);// 订阅这个 LiveDataliveData.observeForever(users-{loadingLiveData.setValue(false);usersLiveData.setValue(users);});// 你也可以直接订阅 Observable并手动设置 LiveData/* model.getUsers() .observeOn(AndroidSchedulers.mainThread()) .subscribe( users - { loadingLiveData.setValue(false); usersLiveData.setValue(users); }, error - { loadingLiveData.setValue(false); errorLiveData.setValue(error.getMessage()); } ); */}OverrideprotectedvoidonCleared(){super.onCleared();// ViewModel 被销毁时LiveDataReactiveStreams 会自动取消订阅// 如果是手动 subscribe需要在此处管理 Disposable}}// 3. View 层 (Activity/Fragment)publicclassUserActivityextendsAppCompatActivity{privateUserViewModelviewModel;OverrideprotectedvoidonCreate(BundlesavedInstanceState){super.onCreate(savedInstanceState);setContentView(R.layout.activity_user);viewModelnewViewModelProvider(this).get(UserViewModel.class);// 观察数据变化viewModel.getUsers().observe(this,users-{// 更新 UI例如设置 Adapter});viewModel.isLoading().observe(this,isLoading-{if(isLoading){// 显示加载框}else{// 隐藏加载框}});viewModel.getError().observe(this,errorMsg-{if(errorMsg!null){// 显示错误信息}});// 触发加载viewModel.loadUsers();}}优势:解耦更彻底: View 通过观察数据变化来更新无需主动调用方法。生命周期感知: LiveData 和 StateFlow 能自动感知 Activity/Fragment 的生命周期避免在非活跃状态下更新 UI。数据驱动 UI: UI 的状态完全由数据决定逻辑更清晰。挑战:桥接成本: 需要将 RxJava 流转换为 LiveData 或 StateFlow增加了复杂性。学习曲线: 需要同时理解 MVVM、数据绑定和 RxJava。1.3. RxBus (事件总线)利用 PublishSubject 或 BehaviorSubject 实现组件间的解耦通信。public class RxBus { private final PublishSubjectObject bus PublishSubject.create(); public void post(Object event) { bus.onNext(event); } public T ObservableT toObservable(ClassT eventType) { return bus.ofType(eventType); } } // 发送事件 RxBus.getInstance().post(new UserLoginEvent(Alice)); // 接收事件 RxBus.getInstance().toObservable(UserLoginEvent.class) .observeOn(AndroidSchedulers.mainThread()) .subscribe(event - { // 更新 UI });1.4. 错误处理与资源管理• 全局错误处理: 使用 RxJavaPlugins.setErrorHandler(…) 设置全局错误处理器。• CompositeDisposable: 管理多个 Disposable在 Activity/Fragment 销毁时统一取消避免内存泄漏。private CompositeDisposable disposables new CompositeDisposable(); // 添加订阅 disposables.add(apiService.getData().subscribe(...)); // 在 onDestroy 中清理 Override protected void onDestroy() { super .onDestroy(); disposables.clear(); // 取消所有订阅 }1.5 复杂任务编排RxJava 通过组合操作符如 zip、merge、concat、combineLatest实现多任务并行或串行执行并合并结果。以下是典型场景与实现1.5.1 并行任务合并zip 操作符场景同时发起多个网络请求如获取用户信息和订单列表待所有请求完成后统一处理结果。ObservableUseruserObservableapi.getUser();ObservableOrderListorderObservableapi.getOrders();Observable.zip(userObservable,orderObservable,(user,orders)-newUserOrderResult(user,orders)).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(result-showUserOrders(result),error-handleError(error));关键点zip 按顺序组合多个 Observable 的数据生成新的数据项。发射数据量以最慢的 Observable 为准超出的数据会被丢弃。1.5.2. 顺序任务合并concat 操作符场景依次执行多个任务如先登录再获取数据前一个任务失败则终止后续任务。实现Observable.concat(api.login(),api.getData()).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(data-processData(data),error-handleError(error));关键点concat 严格按顺序执行前一个 Observable 完成后才会订阅下一个。适合需要严格依赖关系的任务链。1.5.3. 动态数据流合并merge 操作符场景合并多个动态数据源如实时股票价格和用户操作日志不保证顺序。实现ObservableStockPricepriceObservableapi.getStockPrices();ObservableUserActionactionObservableapi.getUserActions();Observable.merge(priceObservable,actionObservable).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(event-logEvent(event),error-handleError(error));关键点merge 合并多个 Observable 的数据按时间顺序发射可能交错。适合无严格顺序要求的实时数据流。1.5.4. 最新数据合并combineLatest 操作符场景当两个输入框内容变化时实时验证表单如密码和确认密码是否一致。实现ObservableStringpasswordObservableRxTextView.textChanges(passwordEditText).skipInitialValue().map(CharSequence::toString);ObservableStringconfirmPasswordObservableRxTextView.textChanges(confirmPasswordEditText).skipInitialValue().map(CharSequence::toString);Observable.combineLatest(passwordObservable,confirmPasswordObservable,(password,confirmPassword)-password.equals(confirmPassword)).subscribe(isValid-showValidationResult(isValid),error-handleError(error));关键点combineLatest 在任意一个源 Observable 发射新数据时组合所有源的最新数据。适合需要基于多个输入实时计算的场景。1.6 UI 事件处理1.6.1 防抖优化搜索输入RxView.clicks(submitButton).throttleFirst(1000,TimeUnit.MILLISECONDS)// 1秒内仅允许一次点击.flatMap(voidEvent-api.submitData(data)).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(success-showSuccess(),error-handleError(error));
