Rxjava是一个异步框架，功能和handler类似，特点是链式调用，逻辑简单，也是一种基于观察者模式的响应式编程框架；
1:观察者模式
2:rxjava异步使用
3:操作符介绍
4:rxjava背压
5:rxjava源码分析
6:rxjava+retrofit
观察者模式：
核心思想：被观察者和观察者通过订阅产生一种关系，当被观察者发生一些改变，通知观察者，观察者对应做出相应的回应。
被观察者（Observable），订阅（subscribe），观察者（Observer）
1:Observable 是RxJava描述的事件流
2:ObservableOnSubscribe 是这个事件流的源头，称之为事件源
3:Observer 是整个事件流的订阅者
rx异步使用
1: 创建被观察者（Observable）
story
调用Observable的create（），传入ObservableOnSubscribe对象，重写ObservableOnSubscribe对象的subscribe（），在subscribe（）中，有一个ObservableEmitter对象，这是一个发射器，调用发射器的onNext()，把被观察者（Observable）的事件发送出去.
2: 创建观察者（Observer）
reader
创建Observer，直接new一个Observer重写他的四个方法，
(1) onSubscribe()：当Observer和Observable订阅的时候调用
(2) onNext()：对Observable中的emitter.onNext（）发射出来的事件进行处理，
(3) onError()：不用多说，坏了
(4) onComplete()：Observable发送来的事件全部处理完成，结束调用，
注意onError()和onComplete()是互斥的，只会调用一个
3:订阅(subscribe)
异步实现：调用Observable的observeOn和subscribeOn设置Observable发射事件和Observer处理事件的线程，
story.observeOn(Schedulers.io());发射事件线程是io线程
story.subscribeOn(AndroidSchedulers.mainThread());处理事件线程是main
rx里的规矩就是被观察者subscribe观察者
rxjava操作符使用
(1)创建操作符
(2)变换操作符
(3)合并操作符
(4)功能操作符
创建操作符
基本创建 create();
快速创建&发送事件:just(),fromArray(),fromIterable(),never(),empty(),error();
延时创建:defer(),timer(),interval(),intervalRange(),range(),rangeLong();
～ just:
Observable.just("1","2","3","4");
~ fromArray:
Integer[] numbers = {1,2,3,4};
Observable.fromArray(numbers);
~ fromIterable:
ArrayList<String> arrayList = new ArrayList<>();
arrayList.add("1");
arrayList.add("2");
arrayList.add("3");
arrayList.add("4");
Observable.fromIterable(arrayList);
never:不发送任何事件
empty:只发送Complete事件，即emitter.complete()
error():发送一个异常，传入error（）中。
延时创建：定时操作&周期性操作
defer：直到有观察者（Observer ）订阅时，才动态创建被观察者对象（Observable） & 发送事件
timer: 延迟指定时间后，发送1个数值0,默认是在新线程中执行
Observable.timer(2, TimeUnit.SECONDS)
本质 = 延迟指定时间后，调用一次 onNext(0);
** interval：**
Observable.interval(3,1,TimeUnit.SECONDS)
变换操作符
map(),flatMap(),ContactMap(),Buffer();
合并操作符
Zip(),merge(),count(),等。
功能操作符
subcribe(),subcribeOn(),observeOn(),delay(),do(),等。
背压
Q:在异步情况中，被观察者发送事件的速率和观察者接收事件的速率不一样，会导致缓冲区溢&oom?
A: 对策： 背压策略（back pressure strategy）————控制事件流速
原理：
(1) 反馈控制：被观察者根据观察者接收事件的能力发送事件
(2) 响应式拉取：根据观察者自身情况接收事件
(3) 缓冲区：对超出缓冲区的事件进行丢弃，覆盖，报错
具体使用：Flowable
在flowable用法中，被观察者变成了Flowable类，观察者变成了Subscriber类,其他用法和规则不变;
1:响应式拉取（控制观察者）
image
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
Log.d("TAG", "发送事件 1");
emitter.onNext(1);
Log.d("TAG", "发送事件 2");
emitter.onNext(2);
Log.d("TAG", "发送事件 3");
emitter.onNext(3);
Log.d("TAG", "发送完成");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR)
.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
.subscribe(new Subscriber<Integer>() {
// 步骤2：创建观察者 = Subscriber & 建立订阅关系
@Override
public void onSubscribe(Subscription s) {
// 对比Observer传入的Disposable参数，Subscriber此处传入的参数 = Subscription
// 相同点：Subscription参数具备Disposable参数的作用，
// 即Disposable.dispose()切断连接, 同样的调用Subscription.cancel()切断连接
// 不同点：Subscription增加了void request(long n)
// 作用：决定观察者能够接收多少个事件
// 如设置了s.request(3)，这就说明观察者能够接收3个事件（多出的事件存放在缓存区）
// 官方默认推荐使用Long.MAX_VALUE，即s.request(Long.MAX_VALUE);
Log.d("TAG", "onSubscribe");
s.request(3);
/**如果在异步的情况中request（）没有参数，则认为观察者不接受事件
* 被观察者可以继续发送事件存到缓存区（缓存区大小=128）
* */
}
@Override
public void onNext(Integer integer) {
Log.d("TAG", "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w("TAG", "onError: ", t);
}
@Override
public void onComplete() {
Log.d("TAG", "onComplete");
}
});
2:反馈控制（控制被观察者）
image
在反馈控制中，同步和异步是不同的，先介绍反馈控制的同步实现方法。
RxJava的可以通过observeOn()（改变观察者的执行线程）与subscribeOn()（改变被观察者的执行线程）来控制，两个操作符的入参都是Scheduler对象.
AndroidSchedulers.mainThread()切换到android的主线程，
Schedulers.io()切换到子线程
subscribeOn在操作链上最好只调用一次，如果多次调用，依然只有第一次生效，observeOn可以在链上调用多次，它主要是用来指定下一个操作在哪一个线程上执行
网友评论