当前位置： 首页 > news >正文

Android RxJava框架分析：它的执行流程是如何的？它的线程是如何切换的？如何自定义RxJava操作符？

news 2025/10/31 22:57:41

目录

  1. RxJava是什么？为什么使用。
  2. RxJava是如何使用的呢？
  3. RxJava如何和Retrofit一起使用。
  4. RxJava源码分析。
  • （1）他执行流程是如何的。
  • （2）map
  • （3）线程的切换。
  1. 如何自定义RxJava操作符？

一、RxJava是什么？为什么使用

RxJava 是一个基于 ​​响应式编程范式​​ 的库，用于通过​​观察者模式​​和​​链式操作符​​，简化异步、事件驱动、多线程数据流处理的开发。

简单来说，RxJava 就像是一个​​“流水线工厂”​​，专门处理需要等待的任务（比如网络请求、数据库查询、复杂计算等）。它能把这些任务串成一条流水线，每个环节处理完数据后，自动传给下一个环节，还能灵活控制任务在哪个线程执行（比如后台线程干活，主线程更新UI）。

1.1 为什么要使用RxJava呢？

接下来，我们看看不使用的问题，以网络请求为例。

需求​​：按顺序做三件事（登录 → 查询订单 → 更新UI）。

​​传统写法​​：​​“回调地狱”​​，层层嵌套，像俄罗斯套娃！

// 伪代码：传统嵌套回调（问题代码）
api.login(new Callback<User>() {@Overridepublic void onSuccess(User user) {api.getOrders(user.getId(), new Callback<List<Order>>() {@Overridepublic void onSuccess(List<Order> orders) {runOnUiThread(() -> {showOrders(orders); // 切主线程更新UI});}@Overridepublic void onFailure(Throwable error) {showError(error); // 每个回调都要处理错误！}});}@Overridepublic void onFailure(Throwable error) {showError(error);}
});

​​问题总结​​：

  • ​​代码缩进成“金字塔”​​，维护困难。
  • ​​重复处理错误​​，每个回调都要写 onFailure。
  • ​​手动切换线程​​（如 runOnUiThread），容易遗漏。

​​RxJava写法​​：

// RxJava 链式调用（解决方案）
api.rxLogin()                   // 1. 登录（被观察者）.flatMap(user -> api.rxGetOrders(user.getId())) // 2. 查询订单（操作符）.observeOn(AndroidSchedulers.mainThread()) // 3. 切到主线程.subscribe(orders -> showOrders(orders), // 4. 观察者消费数据error -> showError(error));   // 统一错误处理！

优势​​：
1️ ​​代码变“直线”​​，逻辑清晰。
2️ ​​统一错误处理​​，一个 onError 搞定所有。
3️ ​​自动线程切换​​，不用写 runOnUiThread。

二、RxJava是如何使用的呢？

（1）添加依赖

implementation 'io.reactivex.rxjava3:rxjava:3.1.8'
implementation 'io.reactivex.rxjava3:rxandroid:3.0.2' // Android 需要

（2）使用

Observable.just("你好").subscribe(new Observer<String>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.d(TAG, "onSubscribe: 订阅开始");}@Overridepublic void onNext(@NonNull String s) {Log.d(TAG, "onNext: 拿到事件"+s);}@Overridepublic void onError(@NonNull Throwable e) {Log.d(TAG, "onError: 错误事件");}@Overridepublic void onComplete() {Log.d(TAG, "onComplete: 事件完成");}});

在这里插入图片描述

解释一下执行流程：

（1）首先我们先需要知道几个角色：观察者( Observer)，被观察者( Observable)。

（2）当被观察者发送出数据(调用just方法)你好的时候，那么观察者就会收到消息(subscribe方法就是订阅)。

（3） subscribe() 将观察者与被观察者连接,触发 Observable 开始发射数据,Observer接收并处理事件（数据、错误、完成信号）。

（4）Observer的 onSubscribe 方法订阅时立即调用​​（最先执行）。 通知观察者订阅已建立。

（5） onNext 方法Observable 发射数据时调用。

（6） onComplete() 方法Observable ​​正常完成数据发射​​后调用 。onError 反之。

三、RxJava如何和Retrofit一起使用

其实就是将Retrofit的响应结果交给RxJava来处理

3.1 发起一个请求

（1）我们需要在Retrofit这里配置 RxJava适配器

public class RetrofitClient {private static final String BASE_URL = "https://www.wanandroid.com/";private static Retrofit retrofit;public static WanAndroidService getService() {if (retrofit == null) {retrofit = new Retrofit.Builder().baseUrl(BASE_URL).addConverterFactory(GsonConverterFactory.create()) // Gson 解析.addCallAdapterFactory(RxJava3CallAdapterFactory.create()) // RxJava 支持.build();}return retrofit.create(WanAndroidService.class);}
}

这行代码的作用是 ​​让 Retrofit 支持返回 RxJava 3 的响应式类型​​（如 Observable、Flowable、Single 等），使得网络请求的结果可以直接通过 RxJava 的流式操作符处理。

（2）在接口这里，我们也是使用Observable来接收。

public interface WanAndroidService {// 示例1：登录接口（POST）@FormUrlEncoded@POST("/user/login")Observable<ApiResponse<User>> login(@Field("username") String username,@Field("password") String password);// 示例2：获取首页文章列表（GET）@GET("/article/list/{page}/json")Observable<ApiResponse<List<Article>>> getHomeArticles(@Path("page") int page);
}

（3）调用接口

private void login(String username, String password) {WanAndroidService service = RetrofitClient.getService();service.login(username, password).subscribeOn(Schedulers.io()) // 在IO线程发起请求.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理结果.subscribe(new Observer<ApiResponse<User>>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {compositeDisposable.add(d); // 统一管理订阅Log.d(TAG, "onSubscribe: ");}@Overridepublic void onNext(@NonNull ApiResponse<User> response) {Log.d(TAG, "onNext: "+response);}@Overridepublic void onError(@NonNull Throwable e) {// 网络错误Log.d(TAG, "onError: ");}@Overridepublic void onComplete() {// 请求完成Log.d(TAG, "onComplete: ");}});
}

执行逻辑：

  1. service.login(username, password)​​：
    通过 Retrofit 定义的接口发起网络请求，返回一个 Observable<ApiResponse<User>>。注意 ​​此时网络请求尚未执行​​，只是定义了数据源。
  2. .subscribeOn(Schedulers.io())​​：
    指定 Observable 的工作线程为 ​​IO 线程​​
  3. observeOn(AndroidSchedulers.mainThread())​​：
    指定 Observer 的回调方法（onNext、onError、onComplete）在 ​​主线程​​ 执行。
  4. ​​.subscribe(Observer)​​：
    订阅 Observable，触发网络请求执行，并绑定观察者处理结果。 ​​此时网络请求正式启动​​。

3.2 发起两个请求，网络嵌套

需求：先登录，登录成功后再获取用户的收藏列表。

// 获取收藏列表 (GET, 需要登录态)
@GET("/lg/collect/list/{page}/json")
Observable<CollectListResponse> getCollectList(@Path("page") int page
);

private void nestedNetworkRequest() {WanAndroidService service = RetrofitClient.getService();service.login("xxx", "xxx").flatMap(loginResponse -> {Log.d(TAG, "nestedNetworkRequest: "+loginResponse);// 登录成功后，获取收藏列表if (loginResponse.getErrorCode() == 0) {return service.getCollectList(0); // 第0页} else {return Observable.error(new Throwable("登录失败"));}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<CollectListResponse>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {}@Overridepublic void onNext(@NonNull CollectListResponse response) {Log.d(TAG, "onNext: "+response);}@Overridepublic void onError(@NonNull Throwable e) {Log.d(TAG, "onError: "+e.getMessage());}@Overridepublic void onComplete() {// 请求完成}});
}

flatMap方法是当 login 请求成功返回数据后才会调用。

在这里插入图片描述

四、源码分析：

4.1 先从Observer(观察者)开始

里面会有一个泛型，onNext也是使用这个泛型。

在这里插入图片描述

4.2 Observable(被观察者)

（1）调用create方法的时候，就会创建一个ObservableCreate
在这里插入图片描述

ObservableCreate里面，包裹了source，按照上面的例子，就类似于调用了我们的login方法，将要发送的请求包裹起来。那么包裹起来干嘛？因为他不是现在执行的，我们都知道，需要调用订阅，才会触发整个流程执行。

4.3 Observable的subscribe订阅过程

在这里插入图片描述

​​订阅发生​​：调用 subscribe() 时，触发 subscribeActual。

在这里插入图片描述

进来以后，我们就可以看到，先执行了我们的onSubscribe方法，然后再去执行source，source就是我们前面说的，将请求包裹起来的内容。

我们看一个非常原始的代码。
在这里插入图片描述

然后再里面调用了onNext，就是执行了观察者的onNext方法，然后执行onCompleter那么到这里，整个执行流程就结束了。

4.4 Map变换操作符原理

为什么map可以改变onnext的接收类型呢？我们继续看看。

在这里插入图片描述

可以看到，这里的类型就进行了转换。但是为什么观察者也跟着变了呢？
在这里插入图片描述

在这里的时候，就已经处理。 返回一个R类型的Observable实例，那么T也就变成了R。

4.5 异步线程切换：subscribeOn(Schedulers.io())

我们拿一个代码来进行分析。

service.login(username, password).subscribeOn(Schedulers.io()) // 在IO线程发起请求.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理结果.subscribe(new Observer<ApiResponse<User>>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {compositeDisposable.add(d); // 统一管理订阅Log.d(TAG, "onSubscribe: ");}...}
}

subscribeOn接收一个Scheduler参数，Scheduler类它封装了线程池和线程切换逻辑。

在这里插入图片描述

在这里插入图片描述

那么Schedulers.io()做了什么？Schedulers.io()​​是一种策略，他会将内部的线程池配置成IO密集型的。我们会发现里面有很多种策略。

在这里插入图片描述

那么我们看看subscribeOn做了什么呢？他拿到线程池以后，做什么呢？先保存起来。我们先记住subscribeOn返回了一个ObservableSubscribeOn类
在这里插入图片描述

因为我们知道任务需要靠订阅方法才触发的，所以我们来看看ObservableSubscribeOn中的
subscribeActual订阅方法
在这里插入图片描述

scheduler.scheduleDirect

在这里插入图片描述

createWorker是一个抽象方法，因为前面我们配置的是Schedulers.io()，所以打开IoScheduler的createWorker
，然后会调用schedult方法执行现成。

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

所以我们的任务就被异步线程执行了。

4.5 主线程切换

那么他如何从异步线程，又切换回主线程的？

.observeOn(AndroidSchedulers.mainThread())

AndroidSchedulers.mainThread()也是一个Scheduler，这里就不多介绍了。我们主要看看他返回的Scheduler

在这里插入图片描述

最终切换主线程，还是使用到了handler

在这里插入图片描述

我们记住HandlerScheduler类

在这里插入图片描述

我们来到observeOn方法

在这里插入图片描述

所以，然后最终交给了HandlerScheduler类来执行。

好了，到这里，源码分析就结束了。

那么我们看源码是为了什么？我们可以自定义RxJava操作符来玩玩，也会让我们对前面学习更加的通透理解。

五、自定义RxJava操作符

我们之定义RxJava操作符，并不是说我们自己实现Observable，而是继承他去实现一些功能。

我们看过just方法就知道，其实继承了Observable
在这里插入图片描述

在这里插入图片描述

然后重写subscribeActual方法，将value包裹进行处理，然后再调用观察者进行分发。

下面我们就写一个防抖操作符，帮助我们理解整个流程。

5.1 自定义防抖操作符

public class RxView {private final static String TAG = RxView.class.getSimpleName();// 我们自己的操作符 == 函数public static Observable<Object> clicks(View view) {return new ViewClickObservable(view);}}


public class ViewClickObservable extends Observable<Object> {private final View view;private static final Object EVENT = new Object();public ViewClickObservable(View view) {this.view = view;}@Overrideprotected void subscribeActual(Observer<? super Object> observer) {MyListener myListener = new MyListener(view, observer);//1.拿到view进行处理observer.onSubscribe(myListener);this.view.setOnClickListener(myListener);}// 拿到viewstatic final class MyListener implements View.OnClickListener, Disposable {private final View view;private Observer<Object> observer;private final AtomicBoolean isDisposable = new AtomicBoolean();public MyListener(View view, Observer<Object> observer) {this.view = view;this.observer = observer;}@Overridepublic void onClick(View v) {if (isDisposed() == false) {observer.onNext(EVENT);}}// 如果用调用了 中断@Overridepublic void dispose() {// 如果没有中断过，才有资格，   取消view.setOnClickListener(null);if (isDisposable.compareAndSet(false, true)) {// 主线程 很好的中断if (Looper.myLooper() == Looper.getMainLooper()) {view.setOnClickListener(null);} else { // 主线程，通过Handler的切换/*new Handler(Looper.getMainLooper()) {@Overridepublic void handleMessage(@NonNull Message msg) {super.handleMessage(msg);view.setOnClickListener(null);}};*///放到主线程执行。AndroidSchedulers.mainThread().scheduleDirect(new Runnable() {@Overridepublic void run() {view.setOnClickListener(null);}});}}}@Overridepublic boolean isDisposed() {return isDisposable.get();}}
}

在这里插入图片描述

RxView.clicks(button).throttleFirst(2000, TimeUnit.MILLISECONDS).subscribe(new Consumer<Object>() {@Overridepublic void accept(Object o) throws Exception {}});}
}

执行流程：

  • 用户通过 RxView.clicks(view) 创建 ViewClickObservable。
  • 调用 subscribe() 后触发 subscribeActual，创建 MyListener 并绑定到 View 的点击事件。
  • 点击事件触发 onClick，通过 Observer 发送 onNext 事件。然后Observer的accept方法就收到了事件(object)
  • 调用 dispose() 时移除 View 的点击监听。
http://www.dtcms.com/a/180862.html

相关文章：

  • 第十七节：图像梯度与边缘检测-Sobel 算子
  • Uskin阵列式三轴力触觉传感器：驱动机器人智能的触觉数据专家
  • 深入理解 Java 代理模式：从基础到实战​
  • MiM: Mask in Mask Self-SupervisedPre-Training for 3D Medical Image Analysis
  • Docker宿主机IP获取
  • 智慧工会服务平台建设方案Word(23页)
  • 机器学习-无量纲化与特征降维（一）
  • 爬虫学习————开始
  • AI服务器通常会运用在哪些场景当中？
  • vue dev-tools插件
  • 电动汽车充换电设施可调能力聚合评估与预测 - 使用说明文档
  • 亚马逊跨境新蓝海：解码爱尔兰电商市场的凯尔特密码
  • HDLC（High-Level Data Link Control，高级数据链路控制协议）
  • uniapp-商城-47-后台 分类数据的生成（通过数据）
  • uniapp 不同路由之间的区别
  • 高频数据结构面试题总结
  • 数据类型：List
  • [特殊字符]适合母亲节的SVG模版[特殊字符]
  • 解决 Redis 缓存与数据库一致性问题的技术指南
  • 配置Java Selenium Web自动化测试环境
  • 在Mathematica中加速绘制图形（LibraryLink）
  • Linux——进程信号
  • 【Qt】编译 Qt 5.15.x For Windows 基础教程 Visual Studio 2019 MSVC142 x64
  • 【什么是 Referrer Policy ？（详解：来源地址策略；访客从哪里来）】
  • project从入门到精通（四）
  • 智芯Z20K144x MCU开发之时钟架构
  • Git_idea界面进行分支合并到主分支详细操作
  • C++编程语言：标准库：标准库概观(Bjarne Stroustrup)
  • 半导体机器人赛道“领跑者”优艾智合：以创新驱动国产化替代加速度
  • Sourcetree安装使用的详细教程