本文共 12707 字,大约阅读时间需要 42 分钟。
Rxjava
由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点,深受各大 Android
开发者的欢迎。Github截图
Rxjava
的初学者Rxjava
入门教程涵盖 基本介绍、原理 & 具体使用等
Rxjava
原理 & 不懂得如何使用的问题希望你们会喜欢。
- 本文主要基于
Rxjava 2.0
- 如果读者还没学习过
Rxjava 1.0
也没关系,因为Rxjava 2.0
只是在Rxjava 1.0
上增加了一些新特性,本质原理 & 使用基本相同- 接下来的时间,我将持续推出
Android
中Rxjava 2.0
的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注!!
示意图
示意图
RxJava
在 GitHub
的介绍:RxJava:a library for composing asynchronous and event-based programs using observable sequences for the Java VM// 翻译:RxJava 是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库
RxJava
是一个 基于事件流、实现异步操作的库实现异步操作
类似于
Android
中的AsyncTask
、Handler
作用
由于 RxJava
的使用方式是:基于事件流的链式调用,所以使得 RxJava
:
更重要的是,随着程序逻辑的复杂性提高,它依然能够保持简洁 & 优雅
Rxjava
原理: 顾客到饭店吃饭示意图
流程图
Rxjava
原理 基于 一种扩展的观察者模式
Rxjava
的扩展观察者模式中有4个角色:
角色 | 作用 | 类比 |
---|---|---|
被观察者(Observable) | 产生事件 | 顾客 |
观察者(Observer) | 接收事件,并给出响应动作 | 厨房 |
订阅(Subscribe) | 连接 被观察者 & 观察者 | 服务员 |
事件(Event) | 被观察者 & 观察者 沟通的载体 | 菜式 |
请结合上述 顾客到饭店吃饭 的生活例子理解:
示意图
流程图
即RxJava
原理可总结为:被观察者 (Observable)
通过 订阅(Subscribe)
按顺序发送事件 给观察者 (Observer)
, 观察者(Observer)
按顺序接收事件 & 作出对应的响应动作。具体如下图:
示意图
至此,RxJava
原理讲解完毕。
RxJava
的基本使用,更深入的RxJava
使用请继续关注Rxjava
的使用方式有两种: Rxjava
的原理 & 使用,主要用于演示说明5.1.1 使用步骤
示意图
5.1.2 步骤详解
步骤1:创建被观察者 (Observable
)& 生产事件
// 1. 创建被观察者 Observable 对象 Observableobservable = Observable.create(new ObservableOnSubscribe () { // create() 是 RxJava 最基本的创造事件序列的方法 // 此处传入了一个 OnSubscribe 对象参数 // 当 Observable 被订阅时,OnSubscribe 的 call() 方法会自动被调用,即事件序列就会依照设定依次被触发 // 即观察者会依次调用对应事件的复写方法从而响应事件 // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递,即观察者模式 // 2. 在复写的subscribe()里定义需要发送的事件 @Override public void subscribe(ObservableEmitter emitter) throws Exception { // 通过 ObservableEmitter类对象产生事件并通知观察者 // ObservableEmitter类介绍 // a. 定义:事件发射器 // b. 作用:定义需要发送的事件 & 向观察者发送事件 emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } });<--扩展:RxJava 提供了其他方法用于 创建被观察者对象Observable -->// 方法1:just(T...):直接将传入的参数依次发送出来 Observable observable = Observable.just("A", "B", "C"); // 将会依次调用: // onNext("A"); // onNext("B"); // onNext("C"); // onCompleted();// 方法2:from(T[]) / from(Iterable ) : 将传入的数组 / Iterable 拆分成具体对象后,依次发送出来 String[] words = {"A", "B", "C"}; Observable observable = Observable.from(words); // 将会依次调用: // onNext("A"); // onNext("B"); // onNext("C"); // onCompleted();
步骤2:创建观察者 (Observer
)并 定义响应事件的行为
Next
事件、Complete
事件 & Error
事件。具体如下:示意图
<--方式1:采用Observer 接口 --> // 1. 创建观察者 (Observer )对象 Observerobserver = new Observer () { // 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件 // 观察者接收事件前,默认最先调用复写 onSubscribe() @Override public void onSubscribe(Disposable d) { Log.d(TAG, "开始采用subscribe连接"); } // 当被观察者生产Next事件 & 观察者接收到时,会调用该复写方法 进行响应 @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件作出响应" + value); } // 当被观察者生产Error事件& 观察者接收到时,会调用该复写方法 进行响应 @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } // 当被观察者生产Complete事件& 观察者接收到时,会调用该复写方法 进行响应 @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } };<--方式2:采用Subscriber 抽象类 -->// 说明:Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类,对 Observer 接口进行了扩展// 1. 创建观察者 (Observer )对象Subscriber subscriber = new Subscriber () {// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件 // 观察者接收事件前,默认最先调用复写 onSubscribe() @Override public void onSubscribe(Subscription s) { Log.d(TAG, "开始采用subscribe连接"); } // 当被观察者生产Next事件 & 观察者接收到时,会调用该复写方法 进行响应 @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件作出响应" + value); } // 当被观察者生产Error事件& 观察者接收到时,会调用该复写方法 进行响应 @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } // 当被观察者生产Complete事件& 观察者接收到时,会调用该复写方法 进行响应 @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } };<--特别注意:2种方法的区别,即Subscriber 抽象类与Observer 接口的区别 -->// 相同点:二者基本使用方式完全一致(实质上,在RxJava的 subscribe 过程中,Observer总是会先被转换成Subscriber再使用)// 不同点:Subscriber抽象类对 Observer 接口进行了扩展,新增了两个方法: // 1. onStart():在还未响应事件前调用,用于做一些初始化工作 // 2. unsubscribe():用于取消订阅。在该方法被调用后,观察者将不再接收 & 响应事件 // 调用该方法前,先使用 isUnsubscribed() 判断状态,确定被观察者Observable是否还持有观察者Subscriber的引用,如果引用不能及时释放,就会出现内存泄露
步骤3:通过订阅(Subscribe
)连接观察者和被观察者
observable.subscribe(observer); // 或者 observable.subscribe(subscriber);
<-- Observable.subscribe(Subscriber) 的内部实现 -->public Subscription subscribe(Subscriber subscriber) { subscriber.onStart(); // 步骤1中 观察者 subscriber抽象类复写的方法,用于初始化工作 onSubscribe.call(subscriber); // 通过该调用,从而回调观察者中的对应方法从而响应被观察者生产的事件 // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递,即观察者模式 // 同时也看出:Observable只是生产事件,真正的发送事件是在它被订阅的时候,即当 subscribe() 方法执行时}
Rxjava
的原理 & 使用RxJava
基于事件流的链式调用// RxJava的链式操作 Observable.create(new ObservableOnSubscribe() { // 1. 创建被观察者 & 生产事件 @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } }).subscribe(new Observer () { // 2. 通过通过订阅(subscribe)连接观察者和被观察者 // 3. 创建观察者 & 定义响应事件的行为 @Override public void onSubscribe(Disposable d) { Log.d(TAG, "开始采用subscribe连接"); } // 默认最先调用复写的 onSubscribe() @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件"+ value +"作出响应" ); } @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } }); }}注:整体方法调用顺序:观察者.onSubscribe()> 被观察者.subscribe()> 观察者.onNext()>观察者.onComplete()
这种 基于事件流的链式调用,使得RxJava
:
更重要的是,随着程序逻辑的复杂性提高,它依然能够保持简洁 & 优雅。所以,一般建议使用这种基于事件流的链式调用方式实现RxJava
。
RxJava 2.x
提供了多个函数式接口 ,用于实现简便式的观察者模式。具体如下:
示意图
以 Consumer
为例:实现简便式的观察者模式
Observable.just("hello").subscribe(new Consumer() { // 每次接收到Observable的事件都会调用Consumer.accept() @Override public void accept(String s) throws Exception { System.out.println(s); } });
我将用一个实际工程实例来演示 Rxjava
的使用
步骤1:加入依赖
compile 'io.reactivex.rxjava2:rxjava:2.0.1' compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
步骤2:直接在MainActivity.java
中实现下述步骤
(Observable )
& 生产事件(Observer )
并 定义响应事件的行为(Subscribe)
连接观察者和被观察者public class MainActivity extends AppCompatActivity { private static final String TAG = "Rxjava"; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main);// 步骤1:创建被观察者 Observable & 生产事件// 即 顾客入饭店 - 坐下餐桌 - 点菜 // 1. 创建被观察者 Observable 对象 Observableobservable = Observable.create(new ObservableOnSubscribe () { // 2. 在复写的subscribe()里定义需要发送的事件 @Override public void subscribe(ObservableEmitter emitter) throws Exception { // 通过 ObservableEmitter类对象产生事件并通知观察者 // ObservableEmitter类介绍 // a. 定义:事件发射器 // b. 作用:定义需要发送的事件 & 向观察者发送事件 emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } });// 步骤2:创建观察者 Observer 并 定义响应事件行为// 即 开厨房 - 确定对应菜式 Observer observer = new Observer () { // 通过复写对应方法来 响应 被观察者 @Override public void onSubscribe(Disposable d) { Log.d(TAG, "开始采用subscribe连接"); } // 默认最先调用复写的 onSubscribe() @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件"+ value +"作出响应" ); } @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } }; // 步骤3:通过订阅(subscribe)连接观察者和被观察者 // 即 顾客找到服务员 - 点菜 - 服务员下单到厨房 - 厨房烹调 observable.subscribe(observer);
示意图
public class MainActivity extends AppCompatActivity { private static final String TAG = "Rxjava"; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main);// RxJava的流式操作 Observable.create(new ObservableOnSubscribe() { // 1. 创建被观察者 & 生产事件 @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } }).subscribe(new Observer () { // 2. 通过通过订阅(subscribe)连接观察者和被观察者 // 3. 创建观察者 & 定义响应事件的行为 @Override public void onSubscribe(Disposable d) { Log.d(TAG, "开始采用subscribe连接"); } // 默认最先调用复写的 onSubscribe() @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件"+ value +"作出响应" ); } @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } }); }}
测试效果
实现效果同上示意图
Demo 下载地址
喜欢的麻烦点个
star
!
public final Disposable subscribe() {} // 表示观察者不对被观察者发送的事件作出任何响应(但被观察者还是可以继续发送事件) public final Disposable subscribe(Consumer onNext) {} // 表示观察者只对被观察者发送的Next事件作出响应 public final Disposable subscribe(Consumer onNext, Consumer onError) {} // 表示观察者只对被观察者发送的Next事件 & Error事件作出响应 public final Disposable subscribe(Consumer onNext, Consumer onError, Action onComplete) {} // 表示观察者只对被观察者发送的Next事件、Error事件 & Complete事件作出响应 public final Disposable subscribe(Consumer onNext, Consumer onError, Action onComplete, Consumer onSubscribe) {} // 表示观察者只对被观察者发送的Next事件、Error事件 、Complete事件 & onSubscribe事件作出响应 public final void subscribe(Observer observer) {} // 表示观察者对被观察者发送的任何事件都作出响应
// 主要在观察者 Observer中 实现 Observerobserver = new Observer () { // 1. 定义Disposable类变量 private Disposable mDisposable; @Override public void onSubscribe(Disposable d) { Log.d(TAG, "开始采用subscribe连接"); // 2. 对Disposable类变量赋值 mDisposable = d; } @Override public void onNext(Integer value) { Log.d(TAG, "对Next事件"+ value +"作出响应" ); if (value == 2) { // 设置在接收到第二个事件后切断观察者和被观察者的连接 mDisposable.dispose(); Log.d(TAG, "已经切断了连接:" + mDisposable.isDisposed()); } } @Override public void onError(Throwable e) { Log.d(TAG, "对Error事件作出响应"); } @Override public void onComplete() { Log.d(TAG, "对Complete事件作出响应"); } };
示意图
Rxjava
的入门知识进行讲解,包括基本介绍、原理 & 具体使用等Android
中 Rxjava 2.0
的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注!!示意图
相关文章阅读
- 操作符使用
- 实际应用讲解