Rxjava---入门篇
2018-08-03 本文已影响6人
一期一会la
1.简介
1.Rxjava是什么：
RxJava 是一个 基于事件流、实现异步操作的库
2.Rxjava的好处：
简洁。随着程序逻辑变得越来越复杂，它依然能够保持简洁。(逻辑复杂，依旧能串成一条线，链式调用)（逻辑简洁,不是指代码量少......）（Android中的AnsynTask与Handler等也是为了让 异步代码更简洁）
3.Rxjava的异步实现：
扩展的观察者模式
2.原理介绍
RxJava原理可总结为：被观察者 （
Observable） 通过 订阅（Subscribe） 按顺序发送事件 给观察者 （Observer）， 观察者（Observer） 按顺序接收事件 & 作出对应的响应动作
| 角色 | 作用 | 类比 |
|---|---|---|
| 被观察者（Observable） | 产生事件 | 顾客 |
| 观察者（Observer） | 接收事件 ，并给出响应动作 | 厨房 |
| 订阅（Subscribe） | 连接 被观察者 & 观察者 | 服务员 |
| 事件（Event） | 被观察者 & 观察者 沟通的载体 | 菜式 |
3.基本使用
- Rxjava的使用方式有两种：
1.分步骤实现：该方法主要为了深入说明Rxjava的原理 & 使用，主要用于演示说明
2.基于事件流的链式调用：主要用于实际使用
- 步骤：
1.创建被观察者（Observable）&生产事件
2.创建观察者（Observer）并定义响应事件的行为
3.通过订阅（Subscribe）连接观察者和被观察者
3.1 步骤详解
1.创建被观察者（Observable）&生产事件:
// 1. 创建被观察者 Observable 对象
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
// create() 是 RxJava 最基本的创造事件序列的方法
// 此处传入了一个 OnSubscribe 对象参数
// 当 Observable 被订阅时，OnSubscribe 的 call() 方法会自动被调用，即事件序列就会依照设定依次被触发
// 即观察者会依次调用对应事件的复写方法从而响应事件
// 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式
// 2. 在复写的subscribe（）里定义需要发送的事件
@Override
public void subscribe(ObservableEmitter<Integer> 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<? extends T>) : 将传入的数组 / Iterable 拆分成具体对象后，依次发送出来
String[] words = {"A", "B", "C"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("A");
// onNext("B");
// onNext("C");
// onCompleted();
2.创建观察者（Observer）并定义响应事件的行为
<--方式1：采用Observer 接口 -->
// 1. 创建观察者 （Observer ）对象
Observer<Integer> observer = new Observer<Integer>() {
// 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<String> subscriber = new Subscriber<Integer>() {
// 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是否还持有观
3.通过订阅（Subscribe）连接观察者和被观察者
observable.subscribe(observer);
// 或者 observable.subscribe(subscriber)；
3.2优雅的实现方法 - 基于事件流的链式调用
整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察
// RxJava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 创建被观察者 & 生产事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
// 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（）>观察
特别注意
RxJava 2.x 提供了多个函数式接口 ，用于实现简便式的观察者模式。具体如下：
示意图
以 Consumer为例：实现简便式的观察者模式:
Observable.just("hello").subscribe(new Consumer<String>() {
// 每次接收到Observable的事件都会调用Consumer.accept（）
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
4.完整实例
4.1 分步骤
加入依赖：
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
public class MainActivity extends AppCompatActivity {
String tag="Rxjava";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
//步骤1:创建被观察者Observable对象
Observable<Integer> observable=Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {//生产事件
Log.d(tag, "被观察者发送事件");
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
}
});
//创建观察者并响应事件行为
Observer<Integer> observer=new Observer<Integer>() {
private Disposable disposable;
@Override
public void onSubscribe(Disposable d) {
Log.d(tag, "开始采用subscribe连接");
disposable=d;
}
@Override
public void onNext(Integer value) {
if(value==2){
disposable.dispose();
Log.d(tag, "已经切断了连接：" + disposable.isDisposed());
}
Log.d(tag, "对Next事件"+ value +"作出响应" );
}
@Override
public void onError(Throwable e) {
Log.d(tag, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(tag, "对Complete事件作出响应");
}
};
//通过订阅（subscribe）连接观察者和被观察者
observable.subscribe(observer);
}
运行结果：
image.png
4.2 基于事件流的链式调用方式
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<Integer>() {
// 1. 创建被观察者 & 生产事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
// 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事件作出响应");
}
});
}
}
image.png
