本文参考
这可能是最好的RxJava 2.x 教程（完结版）
RxJava 2.x 拥有了新的特性，其依赖于4个基础接口，它们分别是
Publisher
Subscriber
Subscription
Processor
其中最核心的莫过于 Publisher 和 Subscriber。Publisher 可以发出一系列的事件，而 Subscriber 负责和处理这些事件。
其中用的比较多的自然是 Publisher 的 Flowable，它支持背压。关于背压给个简洁的定义就是：
背压是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。
简而言之，背压是流速控制的一种策略。有兴趣的可以看一下官方对于背压的讲解。
可以明显地发现，RxJava 2.x 最大的改动就是对于 backpressure 的处理，为此将原来的 Observable 拆分成了新的 Observable 和 Flowable，同时其他相关部分也同时进行了拆分，但令人庆幸的是，是它，是它，还是它，还是我们最熟悉和最喜欢的 RxJava。
观察者模式
大家可能都知道， RxJava 以观察者模式为骨架，在 2.0 中依旧如此。
不过此次更新中，出现了两种观察者模式：
Observable ( 被观察者 ) / Observer ( 观察者 )
Flowable （被观察者）/ Subscriber （观察者）
image.png
在 RxJava 2.x 中，Observable 用于订阅 Observer，不再支持背压（1.x 中可以使用背压策略），而 Flowable 用于订阅 Subscriber ， 是支持背压（Backpressure）的。
image.png
第一步：初始化 Observable
第二步：初始化 Observer
第三步：建立订阅关系
// RxJava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 创建被观察者 & 生产事件
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
}
}).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事件作出响应");
}
});
线程调度
subScribeOn
用于指定发送者发送消息的线程
observeOn
用于指定接受者接送消息的线程
线程切换需要注意的
RxJava 内置的线程调度器的确可以让我们的线程切换得心应手，但其中也有些需要注意的地方。
简单地说，subscribeOn() 指定的就是发射事件的线程，observerOn 指定的就是订阅者接收事件的线程。
多次指定发射事件的线程只有第一次指定的有效，也就是说多次调用 subscribeOn() 只有第一次的有效，其余的会被忽略。
但多次指定订阅者接收线程是可以的，也就是说每调用一次 observerOn()，下游的线程就会切换一次。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
Log.e(TAG, "Observable thread is : " + Thread.currentThread().getName());
e.onNext(1);
e.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "After observeOn(mainThread)，Current thread is " + Thread.currentThread().getName());
}
})
.observeOn(Schedulers.io())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "After observeOn(io)，Current thread is " + Thread.currentThread().getName());
}
});
实例代码中，分别用 Schedulers.newThread() 和 Schedulers.io() 对发射线程进行切换，并采用 observeOn(AndroidSchedulers.mainThread() 和 Schedulers.io() 进行了接收线程的切换。可以看到输出中发射线程仅仅响应了第一个 newThread，但每调用一次 observeOn() ，线程便会切换一次，因此如果我们有类似的需求时，便知道如何处理了。
RxJava 中，已经内置了很多线程选项供我们选择，例如有：
Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作；
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作；
Schedulers.newThread() 代表一个常规的新线程；
AndroidSchedulers.mainThread() 代表Android的主线程
这些内置的 Scheduler 已经足够满足我们开发的需求，因此我们应该使用内置的这些选项，而 RxJava 内部使用的是线程池来维护这些线程，所以效率也比较高。
操作符
Create
create 操作符应该是最常见的操作符了，主要用于产生一个 Obserable 被观察者对象，为了方便大家的认知，以后的教程中统一把被观察者 Observable 称为发射器（上游事件），观察者 Observer 称为接收器（下游事件）。
image.png
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
Log.e(TAG, "Observable emit 1" + "\n");
e.onNext(1);
Log.e(TAG, "Observable emit 2" + "\n");
e.onNext(2);
Log.e(TAG, "Observable emit 3" + "\n");
e.onNext(3);
e.onComplete();
Log.e(TAG, "Observable emit 4" + "\n" );
e.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private int i;
private Disposable mDisposable;
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG, "onSubscribe : " + d.isDisposed() + "\n" );
mDisposable = d;
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e(TAG, "onNext : value : " + integer + "\n" );
i++;
if (i == 2) {
// 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
mDisposable.dispose();
Log.e(TAG, "onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
}
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete" + "\n" );
}
});
需要注意的几点是：
在发射事件中，我们在发射了数值 3 之后，直接调用了 e.onComlete()，虽然无法接收事件，但发送事件还是继续的。
另外一个值得注意的点是，在 RxJava 2.x 中，可以看到发射事件方法相比 1.x 多了一个 throws Excetion，意味着我们做一些特定操作再也不用 try-catch 了。
并且 2.x 中有一个 Disposable 概念，这个东西可以直接调用切断，可以看到，当它的 isDisposed() 返回为 false 的时候，接收器能正常接收事件，但当其为 true 的时候，接收器停止了接收。所以可以通过此参数动态控制接收事件了。
Map
Map 基本算是 RxJava 中一个最简单的操作符了，熟悉 RxJava 1.x 的知道，它的作用是对上游发送的每一个事件应用一个函数，使得每一个事件都按照指定的函数去变化，而在 2.x 中它的作用几乎一致。
image.png
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "this is result " + integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "accept : " + s +"\n" );
}
});
map 基本作用就是将一个 Observable 通过某种函数关系，转换为另一种 Observable，上面例子中就是把我们的 Integer 数据变成了 String 类型。从Log日志显而易见。
Zip
zip 专用于合并事件，该合并不是连接（连接操作符后面会说），合并事件专用,分别从两个上游事件中各取出一个组合,一个事件只能被使用一次，顺序严格按照事件发送的顺序,最终下游事件收到的是和上游事件最少的数目相同（必须两两配对，多余的舍弃)
Observable.zip(getStringObservable(), getInterObservable(), new BiFunction<String, Integer, String>() {
@Override
public String apply(String s, Integer integer) throws Exception {
return s+integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "zip : accept : " + s + "\n");
}
});
private Observable<String> getStringObservable() {
return Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
if (!e.isDisposed()) {
e.onNext("A");
Log.e(TAG, "String emit : A \n");
e.onNext("B");
Log.e(TAG, "String emit : B \n");
e.onNext("C");
Log.e(TAG, "String emit : C \n");
}
}
});
}
private Observable<Integer> getInterObservable() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
if (!e.isDisposed()) {
e.onNext(1);
Log.e(TAG, "Integer emit : 1 \n");
e.onNext(2);
Log.e(TAG, "Integer emit : 2 \n");
e.onNext(3);
Log.e(TAG, "Integer emit : 3 \n");
e.onNext(4);
Log.e(TAG, "Integer emit : 4 \n");
e.onNext(5);
Log.e(TAG, "Integer emit : 5 \n");
}
}
});
}
运行结果.png
zip 组合事件的过程就是分别从发射器 A 和发射器 B 各取出一个事件来组合，并且一个事件只能被使用一次，组合的顺序是严格按照事件发送的顺序来进行的，运行后可以看到，1 永远是和 A 结合的，2 永远是和 B 结合的。
最终接收器收到的事件数量是和发送器发送事件最少的那个发送器的发送事件数目相同，运行后看到，5 很孤单，没有人愿意和它交往，孤独终老的单身狗。
Concat
对于单一的把两个发射器连接成一个发射器，虽然 zip 不能完成，但我们还是可以自力更生，官方提供的 concat 让我们的问题得到了完美解决。
image.png
//两个发射器连接成一个发射器
Observable.concat(Observable.just(1, 2, 3, 4), Observable.just("hello", 5, 6))
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object serializable) throws Exception {
Log.e(TAG, "concat : "+ serializable + "\n" );
}
});
FlatMap
FlatMap将一个发送事件的上游Observable变换成多个发送事件的Observables，然后将它们发射的事件合并后放进一个单独的Observable里，flatMap 并不能保证事件的顺序
image.png
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
int delayTime = (int) (1 + Math.random() * 100);
return Observable.fromIterable(list).delay(delayTime, TimeUnit.MICROSECONDS);
// return Observable.fromIterable(list);
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object s) throws Exception {
Log.e(TAG, "flatMap : accept : " + s + "\n");
}
});
ConcatMap
ConcatMap将一个发送事件的上游Observable变换成多个发送事件Observables，然后将它们发射的事件合并后放进一个单独的Observable里，flatMap 并不能保证事件的顺序，ConcatMap 可以保证事件发送的顺序
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
int delayTime = (int) (1 + Math.random() * 10);
return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
Log.e(TAG, "flatMap : accept : " + o + "\n");
}
});
distinct
去重操作符，其实就是简单的去重
image.png
Observable.just(1, 2, 3, 1, 2, 34, 45, 55)
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "distinct : " + integer + "\n");
}
});
Filter
Filter 过滤操作符。可以接受一个参数，让其过滤掉不符合我们条件的值
image.png
Observable.just(1, 20, 65, -5, 19)
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer >= 10;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "filter : " + integer + "\n");
}
});
buffer
buffer 操作符接受两个参数，buffer(count,skip)，作用是将 Observable 中的数据按 skip (步长) 分成最大不超过 count 的 buffer ，然后生成一个 Observable
![]()
image.png
image.png
Observable.just(1, 2, 3, 4, 5)
.buffer(3,2)
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(List<Integer> integers) throws Exception {
Log.e(TAG, "buffer size : " + integers.size() + "\n");
Log.e(TAG, "buffer value : " + integers.toString());
for (Integer i : integers) {
Log.e(TAG, i + "");
}
Log.e(TAG, "\n");
}
});
timer
在Rxjava中timer操作符既可以延迟执行一段逻辑，也可以间隔执行一段逻辑【注意】但在RxJava2.x已经过时了，现在用interval操作符来间隔执行，但需要注意的是，timer 和 interval 均默认在新线程。
![]()
image.png
image.png
Observable.timer(2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e(TAG, "timer :" + aLong + " at " + DateUtil.getStringDate() + "\n");
}
});
interval
interval 操作符用于间隔时间执行某个操作，其接受三个参数，分别是第一次发送延迟，间隔时间，时间单位。默认在新线程
![]()
image.png
image.png
Disposable mDisposable;
Log.e(TAG, "interval start : " + DateUtil.getStringDate() + "\n");
mDisposable = Observable.interval(3, 2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())// 由于interval默认在新线程，所以我们应该切回主线程
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e(TAG, "interval :" + aLong + " at " + DateUtil.getStringDate() + "\n");
}
});
@Override
protected void onDestroy() {
super.onDestroy();
if (mDisposable != null && !mDisposable.isDisposed()) {
mDisposable.dispose();
}
}
doOnNext
它的作用是让订阅者在接收到数据之前干点有意思的事情。假如我们在获取到数据之前想先保存一下它
Observable.just(1, 2, 3, 4)
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "doOnNext 保存 " + integer + "成功" + "\n");
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "doOnNext :" + integer + "\n");
}
});
skip
skip 很有意思，其实作用就和字面意思一样，接受一个 long 型参数 count ，代表跳过 count 个数目开始接收。
![]()
image.png
image.png
Observable.just(1, 2, 3, 4, 5).skip(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "skip : "+integer + "\n");
}
});
take
take，接受一个 long 型参数 count ，用于指定订阅者最多收到多少数据
image.png
Flowable.fromArray(1, 2, 3, 4, 5)
.take(2)//最多接收多少个参数
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept: take : "+integer + "\n" );
}
});
just
就是一个简单的发射器依次调用 onNext() 方法。
image.png
Observable.just("1", "2",1,5,7)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Serializable>() {
@Override
public void accept(Serializable serializable) throws Exception {
Log.e(TAG,"accept : onNext : " + serializable + "\n" );
}
});
Single
顾名思义，Single 只会接收一个参数，而 SingleObserver 只会调用 onError() 或者 onSuccess()。
Single.just(new Random().nextInt())
.subscribe(new SingleObserver<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onSuccess(Integer value) {
Log.e(TAG, "single : onSuccess : "+value+"\n" );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "single : onError : "+e.getMessage()+"\n");
}
});
debounce
去除发送频率过快的项 debounce(500, TimeUnit.MILLISECONDS) 去除发送间隔时间小于500毫秒的发射事件、或者用于过滤掉发射速率过快的数据项
image.png
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// send events with simulated time wait
emitter.onNext(1); // skip
Thread.sleep(400);
emitter.onNext(2); // deliver
Thread.sleep(505);
emitter.onNext(3); // skip
Thread.sleep(100);
emitter.onNext(4); // deliver
Thread.sleep(605);
emitter.onNext(5); // deliver
Thread.sleep(510);
emitter.onComplete();
}
}).debounce(500, TimeUnit.MICROSECONDS) //去除发送间隔小于500毫秒的发送事件
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "debounce :" + integer + "\n");
}
});
Defer
简单地时候就是每次订阅都会创建一个新的 Observable，并且如果没有被订阅，就不会产生新的 Observable。
![]()
image.png
image.png
Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource<? extends Integer> call() throws Exception {
return Observable.just(1, 2, 3, 4, 5);
}
});
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
Log.e(TAG, "defer : " + value + "\n");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "defer : onError : " + e.getMessage() + "\n");
}
@Override
public void onComplete() {
Log.e(TAG, "defer : onComplete\n");
}
});
last
last 操作符仅取出可观察到的最后一个值，或者是满足某些条件的最后一项。
![]()
image.png
image.png
Observable.just(1, 2, 3,99,33,0)
.last(5)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("RxJavaAct", "last : " + integer + "\n");
}
});
merge
merge 合并，熟悉版本控制工具的你一定不会不知道 merge 命令，而在 Rx 操作符中，merge 的作用是把多个 Observable 结合起来，接受可变参数，也支持迭代器集合。注意它和 concat 的区别在于，不用等到 发射器 A 发送完所有的事件再进行发射器 B 的发送
image.png
Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5, 99))
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("RxJavaAct", "accept : " + integer + "\n");
}
});
reduce
reduce 操作符一次用一个方法处理一个值，可以有一个 seed 作为初始值。
image.png
Observable.just(1, 2, 3)
.reduce(new BiFunction<Integer, Integer, Integer>() {
//我们中间采用 reduce ，支持一个 function 为两数值相加
@Override
public Integer apply(Integer integer, Integer integer2) throws Exception {
Log.e("RxJavaAct", "BiFunction: apply : " + integer + " + " + integer2 + " = " + (integer + integer2) + "\n");
return integer + integer2;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("RxJavaAct", "accept: reduce : " + integer + "\n");
}
});
scan
scan 操作符作用和上面的 reduce 一致，唯一区别是 reduce 是个只追求结果的坏人，而 scan 会始终如一地把每一个步骤都输出。
image.png
Observable.just(1, 2, 3)
.scan(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(Integer integer, Integer integer2) throws Exception {
Log.e("RxJavaAct", "BiFunction: apply : " + integer + " + " + integer2 + " = " + (integer + integer2) + "\n");
return integer + integer2;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("RxJavaAct", "accept: reduce : " + integer + "\n");
}
});
window
按照时间划分窗口，将数据发送给不同的Observable
image.png
Log.e("RxJavaAct", "window\n");
Observable.interval(1, TimeUnit.SECONDS)
.take(15)
.window(3, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Observable<Long>>() {
@Override
public void accept(Observable<Long> longObservable) throws Exception {
Log.e("RxJavaAct", "Sub Divide begin...\n");
longObservable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e("RxJavaAct", "Next:" + aLong + "\n");
}
});
}
});
PublishSubject
onNext() 会通知每个观察者，仅此而已
PublishSubject<Integer> publishSubject = PublishSubject.create();
publishSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "First onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "First onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "First onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "First onComplete!\n");
}
});
publishSubject.onNext(1);
publishSubject.onNext(2);
publishSubject.onNext(3);
publishSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "Second onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "Second onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "Second onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "Second onComplete!\n");
}
});
publishSubject.onNext(4);
publishSubject.onNext(5);
publishSubject.onComplete();
AsyncSubject
在调用 onComplete() 之前，除了 subscribe() 其它的操作都会被缓存，
在调用 onComplete() 之后只有最后一个 onNext() 会生效
AsyncSubject<Integer> asyncSubject = AsyncSubject.create();
asyncSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "First onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "First onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "First onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "First onComplete!\n");
}
});
asyncSubject.onNext(1);
asyncSubject.onNext(2);
asyncSubject.onNext(3);
asyncSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "Second onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "Second onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "Second onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "Second onComplete!\n");
}
});
asyncSubject.onNext(4);
asyncSubject.onNext(5);
asyncSubject.onComplete();
BehaviorSubject
// BehaviorSubject 的最后一次 onNext() 操作会被缓存，
// 然后在 subscribe() 后立刻推给新注册的 Observer
BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create();
behaviorSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "First onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "First onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "First onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "First onComplete!\n");
}
});
behaviorSubject.onNext(1);
behaviorSubject.onNext(2);
behaviorSubject.onNext(3);
behaviorSubject.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "Second onSubscribe :"+d.isDisposed()+"\n");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e("RxJavaAct", "Second onNext value :"+integer + "\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "Second onError:"+e.getMessage()+"\n" );
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "Second onComplete!\n");
}
});
behaviorSubject.onNext(4);
behaviorSubject.onNext(5);
behaviorSubject.onComplete();
Completable
// 只关心结果，也就是说 Completable 是没有 onNext 的，
// 要么成功要么出错，不关心过程，在 subscribe 后的某个时间点返回结果
Log.e("RxJavaAct", "Completable\n");
Completable.timer(1, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new CompletableObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e("RxJavaAct", "onSubscribe : d :" + d.isDisposed() + "\n");
}
@Override
public void onComplete() {
Log.e("RxJavaAct", "onComplete\n");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e("RxJavaAct", "onError :" + e.getMessage() + "\n");
}
});
Flowable
专用于解决被压问题
Flowable.just(1, 2, 3, 4)
//seed 作为初始值
.reduce(100, new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(Integer integer, Integer integer2) throws Exception {
Log.e("RxJavaAct", "reduce :" + integer + " + " + integer2 + " = " + (integer + integer2) + "\n");
return integer + integer2;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("RxJavaAct", "Flowable :" + integer + "\n");
}
});
上边这些操作符的使用场景
采用 OkHttp3 配合 map , doOnNext , 线程切换做简单的网络请求
- 1、通过 Observable.create() 方法，调用 OkHttp 网络请求;
- 2、通过 map 操作符结合 Gson , 将 Response 转换为 bean 类;
- 3、通过 doOnNext() 方法，解析 bean 中的数据，并进行数据库存储等操作;
- 4、调度线程，在子线程进行耗时操作任务，在主线程更新 UI;
- 5、通过 subscribe(),根据请求成功或者失败来更新 UI。
Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(ObservableEmitter<Response> e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, MobileAddress>() {
@Override
public MobileAddress apply(Response response) throws Exception {
Log.e(TAG, "map 线程:" + Thread.currentThread().getName() + "\n");
if (response.isSuccessful()) {
ResponseBody body = response.body();
if (body != null) {
Log.e(TAG, "map:转换前:" + response.body().toString());
return new Gson().fromJson(body.string(), MobileAddress.class);
}
}
return null;
}
}).observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<MobileAddress>() {
@Override
public void accept(MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "doOnNext 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "doOnNext: 保存成功：" + mobileAddress.toString() + "\n");
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<MobileAddress>() {
@Override
public void accept(MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "成功:" + mobileAddress.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "失败：" + throwable.getMessage() + "\n");
}
});
使用框架 rx2-Networking
- 1、通过 Rx2AndroidNetworking 的 get() 方法获取 Observable 对象(已解析);
- 2、调度线程，根据请求结果更新 UI。
- 3、 implementation 'com.amitshekhar.android:rx2-android-networking:1.0.0'
Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.build()
.getObjectObservable(MobileAddress.class)
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<MobileAddress>() {
@Override
public void accept(MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "doOnNext:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "doOnNext:" + mobileAddress.toString() + "\n");
}
}).map(new Function<MobileAddress, MobileAddress.ResultBean>() {
@Override
public MobileAddress.ResultBean apply(MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "\n" );
Log.e(TAG, "map:"+Thread.currentThread().getName()+"\n" );
return mobileAddress.getResult();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<MobileAddress.ResultBean>() {
@Override
public void accept(MobileAddress.ResultBean resultBean) throws Exception {
Log.e(TAG, "subscribe 成功:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "成功:" + resultBean.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 失败:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG, "失败："+ throwable.getMessage()+"\n" );
}
});
zip操作符的使用场景
- 结合多个接口的数据再更新 UI
- zip 操作符可以把多个 Observable 的数据接口成一个数据源再发出去
Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.build()
.getObjectObservable(MobileAddress.class);
Observable<CategoryResult> observable2 = Network.getGankApi()
.getCategoryData("Android", 1, 1);
Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() {
@Override
public String apply(MobileAddress mobileAddress, CategoryResult categoryResult) throws Exception {
return "合并后的数据为：手机归属地："+mobileAddress.getResult().getMobilearea()+"人名："+categoryResult.toString();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "accept: 成功：" + s+"\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "accept: 失败：" + throwable+"\n");
}
});
flatMap 使用场景
- 多个网络请求依次依赖
- 1、注册用户前先通过接口A获取当前用户是否已注册，再通过接口B注册;
- 2、注册后自动登录，先通过注册接口注册用户信息，注册成功后马上调用登录接口进行自动登录。
Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows", 1 + "")
.build()
.getObjectObservable(FoodList.class) // 发起获取食品列表的请求，并解析到FootList
.subscribeOn(Schedulers.io()) // 在io线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理获取食品列表的请求结果
.doOnNext(new Consumer<FoodList>() {
@Override
public void accept(@NonNull FoodList foodList) throws Exception {
// 先根据获取食品列表的响应结果做一些操作
Log.e(TAG, "accept: doOnNext :" + foodList.toString());
}
})
.observeOn(Schedulers.io()) // 回到 io 线程去处理获取食品详情的请求
.flatMap(new Function<FoodList, ObservableSource<FoodDetail>>() {
@Override
public ObservableSource<FoodDetail> apply(@NonNull FoodList foodList) throws Exception {
if (foodList != null && foodList.getTngou() != null && foodList.getTngou().size() > 0) {
return Rx2AndroidNetworking.post("http://www.tngou.net/api/food/show")
.addBodyParameter("id", foodList.getTngou().get(0).getId() + "")
.build()
.getObjectObservable(FoodDetail.class);
}
return null;
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodDetail>() {
@Override
public void accept(@NonNull FoodDetail foodDetail) throws Exception {
Log.e(TAG, "accept: success ：" + foodDetail.toString());
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "accept: error :" + throwable.getMessage());
}
});
concat 使用场景
- 先读取缓存数据再读取网络请求
- 实用场景中经常会用到缓存数据，以通过减少频繁的网络请求达到节约流量:
- 1、concat 可以做到不交错的发射两个甚至多个 Observable 的发射物;
- 2、并且只有前一个 Observable 终止（onComplete）才会订阅下一个 Observable
// 先读取缓存再读取网络
Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
@Override
public void subscribe(ObservableEmitter<FoodList> e) throws Exception {
Log.e(TAG, "create当前线程:" + Thread.currentThread().getName());
FoodList data = CacheManager.getInstance().getFoodListJsonData();
// 在操作符concat 中，只有调用onComplete 之后才会执行下一个Observable
if (data != null) {//如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
isFromNet = false;
Log.e(TAG, "\nsubscribe: 读取缓存数据:");
runOnUiThread(new Runnable() {
@Override
public void run() {
Log.e(TAG, "正在 : 读取缓存数据:");
}
});
e.onNext(data);
} else {
isFromNet = true;
runOnUiThread(new Runnable() {
@Override
public void run() {
Log.e(TAG, "正在 : 读取网络数据:");
}
});
Log.e(TAG, "\nsubscribe: 读取网络数据:");
e.onComplete();
}
}
});
// 请求网络数据
Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows", 10 + "")
.build()
.getObjectObservable(FoodList.class);
// Concat 先读取缓存数据并展示UI再获取网络数据刷新UI
// 1、concat 可以做到不交错的发射两个甚至多个 Observable 的发射物;
// 2、并且只有前一个 Observable 终止（onComplete）才会订阅下一个 Observable
// 两个 Observable 的泛型应当保持一致
Observable.concat(cache, network)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodList>() {
@Override
public void accept(FoodList foodList) throws Exception {
Log.e(TAG, "subscribe 成功:" + Thread.currentThread().getName());
if (isFromNet) {
Log.e(TAG, "accept : 网络获取数据设置缓存: \n" + foodList.toString());
CacheManager.getInstance().setFoodListJsonData(foodList);
}
Log.e(TAG, "accept: 读取数据成功:" + foodList.toString());
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 失败:" + Thread.currentThread().getName());
Log.e(TAG, "accept: 读取数据失败：" + throwable.getMessage());
}
});
debounce 使用场景
- 减少频繁的网络请求
- 设想情景：输入框数据变化或者点击一次按钮时就要进行网络请求，这样会产生大量的网络请求，而实际上又不需要
- 这时候可以通过 debounce 过滤掉发射频率过快的请求。
btn_debounce = findViewById(R.id.btn_debounce);
// 减少频繁的网络请求
RxView.clicks(btn_debounce)
.debounce(2, TimeUnit.SECONDS) // 过滤掉发射频率小于2秒的发射事件
.subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
clickBtn();
}
});
private void clickBtn() {
Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows", 2 + "") // 只获取两条数据
.build()
.getObjectObservable(FoodList.class)
.subscribeOn(Schedulers.io()) // 在 io 线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) // 在主线程进行更新UI等操作
.subscribe(new Consumer<FoodList>() {
@Override
public void accept(@NonNull FoodList foodList) throws Exception {
Log.e(TAG, "accept: 获取数据成功:" + foodList.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "accept: 获取数据失败：" + throwable.getMessage() + "\n");
}
});
}
interval 使用场景
- 间隔任务实现心跳
- 可能我们会遇上各种即时通讯，如果是自己家开发的 IM 即时通讯，我相信在移动端一定少不了心跳包的管理
- 而我们 RxJava 2.x 的 interval 操作符棒我们解决了这个问题。
private Disposable mDisposable;
// 间隔任务实现心跳
mDisposable = Flowable
.interval(1, TimeUnit.SECONDS)
.doOnNext(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e(TAG, "accept: doOnNext : " + aLong);
}
}).observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.e(TAG, "accept: 设置文本 ：" + aLong);
}
});
@Override
protected void onDestroy() {
super.onDestroy();
if (mDisposable != null) {
mDisposable.dispose();
}
}
线程调度
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.e(TAG, "Observable thread is : " + Thread.currentThread().getName());
e.onNext(1);
e.onComplete();
}
}).subscribeOn(Schedulers.newThread())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "After observeOn(mainThread)，Current thread is " + Thread.currentThread().getName());
}
})
.observeOn(Schedulers.io())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "After observeOn(io)，Current thread is " + Thread.currentThread().getName());
}
});
或许上边的代码接口有些调用不了
