RxJava2.0初学总结笔记

(个人知识总结，学习RxJava的时候主要是参照作者Carson_Ho的简书)
一.RxJava的作用
RxJava主要是基于事件流，实现异步操作的一个库，链式调用。且RxJava采用的是Java中很出名的“观察者模式”。（在学习了RxJava之后，大家除了什么“单例模式”之外，又懂一种新的模式了---观察者模式。）
类似于Handler，AsyncTask。
二.使用RxJava的优点
个人体会还是主要是使用RxJava会使逻辑清晰。
三.RxJava主要的几个类
1.Observable（被观察者）
2.Observe（观察者）
3.Suscribe（订阅）
三个类的事件传递方向为:Observable(产生事件)---subscribe（将被观察者和观察者进行绑定）--->Observe（接收处理事件）
通俗可以理解为：（大哥）Observe，（小弟）Observable()，被观察者(小弟Observable)一有事，就要用电话（subscribe）给大哥Observe说。大哥接收到小弟（Observable）发出来的讯息，然后开始处理小弟（Observable所发的信息。（个人理解，若比喻有错误，请指正）
既然通俗的理解了，在项目中大概哪些地方才能用到它呢？
1.很多使用是用RxJava配合Retrofit来搭建网络框架。（也是很多开发者最主要使用的方式）
2.代替Handler，AsyncTask。(对于技术不上不下的人来说，用原生AsyncTask来处理异步操作真的要命）。
其他我还没发现到。
四.如何初步的使用RxJava
想要使用首先肯定是要创建好我们的---小弟（[b]Observable）,大哥（Observe），还有我们的电话（Suscribe），然后再把大哥的心，小弟的心，手机的心串一串~串一个链式调用~（玩笑一下）[/b]
1.创建小弟-->[b]Observable[/b]
create方法创建是一种很基础简单的方式。

Observable<String> observableDemo=Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onError(new Exception("发生错误了"));
observableEmitter.onComplete();
}
});

除此之外，还有一种用just（）来快速创建被观察者，由代码可以看出这种方式很明了简洁。Observable observable2 = Observable.just("A", "B", "C");fromXXX();//在2.0以前可以直接用from来调用，但是在2.0之后，我们需要确定你的数据类型，指明你的数据类型是fromArray还是fromIterator...String[] testFrom = {"1", "2"};
//被观察者
Observable observable = Observable.fromArray(testFrom);from的类型一般如下所示




2.创建大哥-->Observer
这种是最基本的创建方式//观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable disposable) {
}

@Override
public void onNext(String s) {
}

@Override
public void onError(Throwable throwable) {
}

@Override
public void onComplete() {
}
};还有一种是由Subscriber来创建

Subscriber subscriber=new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
}

@Override
public void onNext(Object o) {
}

@Override
public void onError(Throwable t) {
}

@Override
public void onComplete() {
}
};

相对于最基本的创建方式来说。subscriber新增了onStart()，unsubscribe（）方法。
onStart():在未响应事件前调用，可以用来做一些准备工作
unsubscribe();取消订阅，当调用这个方法之后，观察者将不再接收和响应事件。
*在调用unsubscribe()方法之前，需要用isunsubscribe()判断状态，判断被观察者（Observable）是否还持有观察者（Observer）Subscriber的引用，如果引用不能及时释放，就会出现内存泄漏。
3.订阅（subscribe）---->电话observable.subscribe(observer);//被观察者，订阅，观察者4.链式创建
一开始就说RxJava是基于事件流的链式调用，现在把它们链接起来看一下。Observable.create(new ObservableOnSubscribe<Integer>() {//创建被观察者
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> observableEmitter) throws Exception {
observableEmitter.onNext(1);
observableEmitter.onNext(2);
observableEmitter.onComplete();
}
}).subscribe(//绑定
new Observer<Integer>() {//创建观察者
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Integer integer) {
Log.i("t","被观察者所发出的值="+integer);
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});这种链式的好处就是使逻辑上变得简单清晰。这也是RxJava的一个优点。
5.切断小弟（Observable）和大哥（Observer）的链接---->Disposable.dispose();Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
emitter.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDisposable;
private int i;

@Override
public void onSubscribe(Disposable d) {
mDisposable = d;
}

@Override
public void onNext(Integer value) {
Log.i("t", "onNext: " + value);
i++;
if (i == 2) {
Log.i("t", "dispose");
mDisposable.dispose();
Log.i("t", "isDisposed : " + mDisposable.isDisposed());//若已切断，isDisposed=true;
}
}

@Override
public void onError(Throwable e) {
}

@Override
public void onComplete() {
}
});五.一些常见常用操作符介绍
just();//快速创建被观察者对象。最多只能发送10个事件。Observable observable2 = Observable.just("A", "B", "C");类似于:onNext("A");onNext("B");onNext("C");
fromArray();//快速创建被观察者对象。发送10个以上的事件。数组形式
fromIterable();//快速创建被观察者对象。发送10个以上的事件。集合形式
观察者接收到的数据为遍历数据。
empty();//仅发送complete事件，直接通知完成
error();//仅发送error事件，直接通知异常
never();//不发送任何事件
-----------------------------------延迟操作符---------------------------------------
defer();//快速创建被观察者对象。此方法作用：直到有观察者订阅才动态创建被观察者对象，并发送事件。每次订阅后都会得到一个新的被观察者(Observable),可以确保Observable的数据是最新的。
//延时操作.有观察者订阅才能动态创建被观察者
Observable<Integer> mobservable=Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource call() throws Exception {
return Observable.just(i);
}
});

i=15;
//观察者被订阅，开始动态创建被观察者
mobservable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Integer integer) {
//i=15
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});timer();//快速创建被观察者对象。方法作用：延迟指定时间后，发送一个onNext(0);的事件。（Long类型）
*timer默认运行在新线程，可用scheduler来指定线程。

//timer(延迟时间,时间单位)
Observable.timer(5, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Long aLong) {
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});interval();//快速创建被观察者对象。每隔指定时间就发送事件。  
//1第一次延迟的时间5s 2执行事件，每隔1s发送一次。 3计数单位为秒
Observable.interval(5,1,TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Long aLong) {
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});发送事件序列：从0开始，五险递增1的整数序列。
intervalRange();//快速创建被观察者对象。每隔一段时间就发送事件，可指定事件数量。
//1.起始数据 2.事件数量 3.第一次延迟时间 4.执行的间隔秒数 5.计数单位
Observable.intervalRange(2,5,3,1,TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {

}

@Override
public void onNext(@NonNull Long aLong) {
//事件数据=2,3,4,5,6
}

@Override
public void onError(@NonNull Throwable throwable) {

}

@Override
public void onComplete() {

}
});range();//快速创建被观察者对象。类似于intervalRange(),但是range()无延迟发送事件。
//1.事件序列起点=0 2.指定事件数量=3
Observable.range(0,3).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Integer integer) {
//收到的事件=0,1,2
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});rangeLong();//快速创建被观察者对象。类似于range()，区别是，rangeLong()支持Long类数据。
------------------------------------变换操作符--------------------------------------
map();//将被观察者发送的事件转换为任意的类型的事件
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
//事件传递为string="1"
observableEmitter.onNext("1");
}
//用map操作Function来对被观察者所发出的事件类型进行变换。string->integer
}).map(new Function<String, Integer>() {
@Override
public Integer apply(@NonNull String s) throws Exception {
return null;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
//接收到的integer=1
}
});FlatMap();//将被观察者发送的事件序列进行拆分转换，再合并为一个新的事件序列，最后再发送。（类似于打乱事件顺序）
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> observableEmitter) throws Exception {
observableEmitter.onNext(100);
observableEmitter.onNext(101);
observableEmitter.onNext(102);
}
//由floatMap把原来的事件生产序列进行拆分，再将每个事件拆分转换成新的,发送三个string最后合并，再发送给被观察者
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
List<String> list=new ArrayList<String>();
for(int i=0;i<3;i++){
list.add("原事件="+integer+"拆分的新事件="+i);
}
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {

}
});*新事件的合成顺序与旧事件的顺序无关。
concatMap();//类似于flatmap()，用法和flatmap差不多.区别是新事件的顺序=被观察者旧序列生产的顺序。有序的变换事件序列。
Buffer();//定期从被观察者中获取一定数量的事件，并放到缓存区中。然后发送。用于缓存被观察者中的事件。
//buffer->1.设置缓存区的大小 2.设置每次获取新事件的数量
//例如第一次是3,4,5被存入，第二次是2,3,4（新事物增加为1，向指针移动一位），一直到null，null，1终止存储
Observable.just(1,2,3,4,5).buffer(3,1).subscribe(new Observer<List<Integer>>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {

}

@Override
public void onNext(@NonNull List<Integer> integers) {

}

@Override
public void onError(@NonNull Throwable throwable) {

}

@Override
public void onComplete() {

}
});----------------------------------组合/合并操作符-----------------------------------concat:组合多个被观察者一起发送数据，合并后按发送顺序串执行。组合被观察者数量<=4
concatArray:组合多个被观察者一起发送数据，合并后按发送顺序串执行。组合被观察者数量>4
concat和concatArray的用法类似，只是数量不一样而已。
Observable.concat(Observable.just(1,2,3),
Observable.just(10,11,12)).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Integer integer) {
//接收到的事件=1,2,3,10,11,12
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});merge()/mergeArray():组合多个被观察者一起发送数据，合并后按时间线并行执行。
两者的区别：merge组合被观察者数量<=4      mergeArray>4(两者用法相似，只是数量不同)
//1.从0开始发送，共发2个数据，第一次事件延迟2s，间隔3s
//2.从1开始发送，共发送3个数据，第一次事件延迟2s，间隔3s
Observable.merge(Observable.intervalRange(0,2,2,3,TimeUnit.SECONDS),
Observable.intervalRange(1,3,2,3,TimeUnit.SECONDS)).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {

}

@Override
public void onNext(@NonNull Long aLong) {
//接收到的数据0,1--->1,2--->3
}

@Override
public void onError(@NonNull Throwable throwable) {

}

@Override
public void onComplete() {

}
});concatDelayError()/mergeDelayError();//多个被观察者中的某一个抛出onError，正常执行是其他的被观察者就不会执行了。但是concatDelay和mergeDelayError则可以让所有的被观察者都发送事件结束之后再触发onError。（两者使用相似，用相应的方式即可）
Observable.concatArrayDelayError(
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
// 发送Error事件，因为使用了concatDelayError，所以第2个Observable将会发送事件，等发送完毕后，再发送错误事件
emitter.onError(new NullPointerException());
emitter.onComplete();
}
}),
Observable.just(4, 5, 6))
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
}

@Override
public void onError(Throwable e) {
}

@Override
public void onComplete() {
}
});Zip();//合并多个被观察者发送的事件，生成一个新的事件序列，然后发送。
*最终合并事件数量=多个被观察者中数量最少的数、
Observable.zip(Observable.just(1, 2, 3,9), Observable.just(10, 11, 12),
//1.被观察者对象1的类型 2.被观察者2的类型 3合并的类型
//Observable的9最终也会被发送出来，但是如果在1,2被观察者事件序列之后complete,则不会发送9
new BiFunction<Integer, Integer, String>() {
@Override
public String apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
return integer+integer2+"";
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull String s) {
//最终接收到的=110,211,312
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});一般应用于，展示的信息从多个地方获取并且统一结合后展示。
网络请求发送和结果的统一显示。（Retrofit+RxJava）
combineLatest();//按事件合并，在同一个事件点上合并  两个被观察者对象中，任何一个先发送事件的最后一个数据和另外一个被观察者对象的每一个数据结合。然后发送。
Observable.combineLatest(Observable.just(1L, 2L, 3L),
//0开始，3个数，第一次延迟1s，间隔1s
Observable.intervalRange(5,3,1,1,TimeUnit.SECONDS),
new BiFunction<Long, Long, Long>() {
@Override
public Long apply(@NonNull Long integer, @NonNull Long integer2) throws Exception {
// integer3 integer2=5; integer=3 integer2=6; integer=3 integer2=7;
return integer+integer2;
}
}).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Long integer) {
//结果=8,9,10
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {
}
});combineLatestDelaError();//和concatDelay(),mergeDelay();用法意义都是一样的。
reduce();//把被观察者中的事件聚合成一个事件，并发送。
Observable.just(1,2,3,4).reduce(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
//第一次是1+2=3，返回数字3，第二次，3+3=6，用返回数加下一个数，第三次6+4=10用返回数加下一个数4.
return integer+integer2;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
//integer=3,6,10
}
});collect();//把被观察者的事件数据收集到一个数据结构里。
Observable.just(1,2,3,4).collect(
//创建数据结构（容器），用于收集被观察者发送的数据
new Callable<ArrayList<Integer>>() {
@Override
public ArrayList<Integer> call() throws Exception {
return new ArrayList<>();
}
//对发送的数据进行收集
}, new BiConsumer<ArrayList<Integer>, Integer>() {
@Override
public void accept(ArrayList<Integer> integers, Integer integer) throws Exception {
integers.add(integer);//数据收集
}
}).subscribe(new Consumer<ArrayList<Integer>>() {
@Override
public void accept(ArrayList<Integer> integers) throws Exception {
//接收到的数组integers：[1,2,3,4]
}
});startWith()/startWithArray();//发送事件前，追加发生事件。
在被观察者在发送事件前，追加发送一个（多个）数据/一个新的Observable（被观察者）
Observable.just(1,2,3)
.startWith(7)//追加一个数据
.startWithArray(8,9,10)//追加多个数据
.startWith(Observable.just(11,12))//追加被观察者
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}

@Override
public void onNext(@NonNull Integer integer) {
}

@Override
public void onError(@NonNull Throwable throwable) {
}

@Override
public void onComplete() {

}
});count();//统计被观察者发送事件的数量
Observable.just(1,2,3,4,5,6).count().subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
//发送事件数量along=6
}
});为了清晰自己的认知，盗用一张Carson_Ho大神的讲解图，超级全面！强推他的RxJava讲解，https://www.jianshu.com/p/cd984dd5aae8



此为个人学习笔记，主要参照Carson_Ho大神的简书文章来总结出来自己习惯看的模式的样子。要是我理解有误的地方，麻烦各位多多指点纠正！