RxJava的学习笔记（二）操作符
2017-08-12 本文已影响0人
jltxseo
前言
RxJava的学习笔记（一）基础概念
RxJava的学习笔记（二）操作符
RxJava的学习笔记（三）线程调度
上一节笔记中记录了RxJava2的基础和基本使用，在本节中继续学习RxJava2的操作符使用。
1、创建操作
创建Observable的方法：
- just( )， 将一个或多个对象转换成发射这个或这些对象的一个Observable
- fromArray( )，将一个Iterable, 一个Future, 或者一个数组转换成一个Observable
- repeat( )， 创建一个重复发射指定数据或数据序列的Observable
- repeatWhen( )，创建一个重复发射指定数据或数据序列的Observable，它依赖于另一个-
Observable发射的数据 - create( )，使用一个函数从头创建一个Observable
- defer( )，只有当订阅者订阅才创建Observable；为每个订阅创建一个新的Observable
- range( )，创建一个发射指定范围的整数序列的Observable
- interval( )，创建一个按照给定的时间间隔发射整数序列的Observable
- timer( )，创建一个在给定的延时之后发射单个数据的Observable
- empty( )，创建一个什么都不做直接通知完成的Observable
- error( )，创建一个什么都不做直接通知错误的Observable
- never( )，创建一个不发射任何数据的Observable
方法虽然很多，具体以页面展示的功能需求为准，这里举例常用的just,from,fromIterable着几个创建操作符:
just直接传入一个或者多个对象，发射器依次发射传入的对象
public void justOperator() {
Observable.just("map", "map1", "map2")
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println("accept:" + s);
}
});
}
fromArray传入一个对象数组
public void fromArrayOperator() {
Observable.fromArray(new String[]{"map", "map1", "map2"})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println("accept:" + s);
}
});
}
fromIterable接收一个 Iterable 容器，例如List
public void fromIterableOperator() {
List<Integer> list = new ArrayList<>();
list.add(10);
list.add(1);
list.add(20);
Observable.fromIterable(list)
.subscribe(num -> System.out.println("accept:" + num));
}
注意:下游观察者仅仅只是关心onNext方法，可以在订阅的时候传入Consumer示例：
//仅仅关心onNext方法。所以可以简写成下面这样
Consumer consumer = new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(XqTag.TAG,"accept:" + s);
}
};
2、变换操作
RxJava的变换操作符有很多，所谓变换，就是将事件序列中的对象或者整个序列进行一定的加工处理，转换成不同的事件或者事件序列。这里演示常用map()、flatMap()、concatMap()、flatMapIterable()这些变换操作符:
map作用是对于上游发射过来的每一个事件都按照指定的函数进行变换，并将结果依次按照同样的序列发射下游。
map.png
示例:
public static void mapOperator() {
Observable.just("map", "map1", "map2")
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return s + "-jltx";
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(XqTag.TAG,"accept:" + s);
}
});
}
变换后的结果:
08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map-jltx
08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map1-jltx
08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map2-jltx
flatMap作用是可以把一个上游发射过来的一个发射器Observable通过某种方法变换成多个Observables发射器，然后依次将他们所有发射的数据发射到下游的观察者。有一点需要注意它并不能保证事件的顺序。
flatMap.png
示例:
public static void flatMapOperator() {
List<Integer> list1 = new ArrayList<>();
list1.add(1);
list1.add(2);
list1.add(3);
List<Integer> list2 = new ArrayList<>();
list2.add(4);
list2.add(5);
list2.add(6);
List<Integer> list3 = new ArrayList<>();
list3.add(7);
list3.add(8);
list3.add(9);
List<Integer> list4 = new ArrayList<>();
list4.add(10);
list4.add(11);
list4.add(12);
Observable.just(list1,list2,list3,list4)
.flatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
//使用一个随机的演示验证事件的顺序
int delayTime = (int) (1 + Math.random() * 20);
return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integers) throws Exception {
Log.d(XqTag.TAG,"accept:" + integers);
}
});
}
变换后的结果:
08-11 12:51:39.928 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 12:51:39.936 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 12:51:39.937 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 12:51:39.939 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 12:51:39.940 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 12:51:39.940 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 12:51:39.940 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 12:51:39.943 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 12:51:39.943 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 12:51:39.944 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:12
08-11 12:51:39.944 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 12:51:39.948 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 12:51:39.948 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:9
如上面通过延时delay一个随机事件得到的结果，验证了flatMap产生多个Observable最后合并的最终发射的数据是无序的。
concatMap作用和flatMap的作用一样，都是一个发射器通过某种方法变换成多个Observables发射器,然后依次将他们所有发射的数据发射到下游的观察者。它和flatMap唯一的区别是它自己保证了顺序。
concatMap.png
示例:
public static void concatMapOperator() {
List<Integer> list1 = new ArrayList<>();
list1.add(1);
list1.add(2);
list1.add(3);
List<Integer> list2 = new ArrayList<>();
list2.add(4);
list2.add(5);
list2.add(6);
List<Integer> list3 = new ArrayList<>();
list3.add(7);
list3.add(8);
list3.add(9);
List<Integer> list4 = new ArrayList<>();
list4.add(10);
list4.add(11);
list4.add(12);
Observable.just(list1,list2,list3,list4)
.concatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
//使用一个随机的演示验证事件的顺序
int delayTime = (int) (1 + Math.random() * 20);
return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integers) throws Exception {
Log.d(XqTag.TAG,"accept:" + integers);
}
});
}
变换后的结果:
08-11 13:01:51.461 4319-4319/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 13:01:51.480 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:9
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:12
由以上结果可知cancatMap可以保证顺序
flatMapIterable作用是变换成对的打包数据，然后生成Iterable而不是原始数据和生成的Observables，但是处理方式是相同的。
示例:
public static void flatMapIterableOperator() {
List<Integer> list1 = new ArrayList<>();
list1.add(1);
list1.add(2);
list1.add(3);
List<Integer> list2 = new ArrayList<>();
list2.add(4);
list2.add(5);
list2.add(6);
List<Integer> list3 = new ArrayList<>();
list3.add(7);
list3.add(8);
list3.add(9);
List<Integer> list4 = new ArrayList<>();
list4.add(10);
list4.add(11);
list4.add(12);
List<List<Integer>> list = new ArrayList<>();
list.add(list1);
list.add(list2);
list.add(list3);
list.add(list4);
Observable.fromIterable(list)
.flatMapIterable(new Function<List<Integer>, Iterable<Integer>>() {
@Override
public Iterable<Integer> apply(List<Integer> integers) throws Exception {
Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
//使用一个随机的演示验证事件的顺序
return integers;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(XqTag.TAG,"accept:" + integer);
}
});
}
得到的结果是:
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:9
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:12
3、过滤操作
过滤操作符可用于过滤和选择Observable发射的数据序列。这里演示常用的filter()、take()这几个过滤操作符：
filter作用是过滤数据的，返回false表示拦截此数据。
示例：
/**
* filter 是用于过滤数据的，返回false表示拦截此数据。
*/
public static void filterOperator() {
Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer.intValue() > 5;
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(XqTag.TAG, "accept:" + integer);
}
});
}
上面的过滤出数据大于5的所有数据
得到的结果:
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:20
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:8
take作用是修改上游Observable的行为，只返回前面的N项数据，然后发射完成通知，忽略剩余的数据。也就是只发射发射前面的N项数据。就算上游Observalbe发射的数据少于N项，那么最后下游接收到的数据个数一样跟上游的个数一样，同事不会抛异常，发射完所有的数据后调用onComplete。
示例:
public static void takeOperator() {
Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
.take(10)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(XqTag.TAG, "onSubscribe");
}
@Override
public void onNext(Integer value) {
Log.d(XqTag.TAG, "onNext:" + value);
}
@Override
public void onError(Throwable e) {
Log.d(XqTag.TAG, "onError:" + e.toString());
}
@Override
public void onComplete() {
Log.d(XqTag.TAG, "onComplete");
}
});
}
得到的结果:
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:10
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:6
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:20
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:-1
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:8
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onComplete
4、组合操作
组合操作一般用于把多个Obserbable按照一定的关系给连接起来，最后在一个公共的订阅者里面进行汇总监听。
merge顾名思义作用是把多个Observalbe提交的结果按照结果的时间顺序结合。一旦某一个的Observable中出现错误就会马上停止合并，并对下游的订阅者执行onError方法。
merge.png
这里使用两个Observable来进行测试，一个被订阅者每隔1000毫秒后开始发射数据，并只取前面的5个数据，它的发射数据顺序为 0 2 4 6 8；另一个被订阅者延时500毫秒后也是每隔1000毫秒发射设局，也是只取前面的5个数据，它的发射数据顺序为1 3 5 7 9 示例:
/**
* 按照两个Observable提交结果的时间顺序，对Observable进行合并
*/
public static void mergeOperator() {
//被订阅者每隔1000毫秒后开始发射数据，并只取前面5个数据 发射顺序为0 2 4 6 8
Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2;
return result;
}
}).take(5);
//被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2 + 1;
return result;
}
}).take(5);
Observable.merge(observable1, observable2).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(XqTag.TAG, "onSubscribe");
}
@Override
public void onNext(Long value) {
Log.d(XqTag.TAG, "onNext:" + value);
}
@Override
public void onError(Throwable e) {
Log.d(XqTag.TAG, "onError:" + e.toString());
}
@Override
public void onComplete() {
Log.d(XqTag.TAG, "onComplete");
}
});
}
得到的结果是：
08-14 10:18:28.657 9735-9735/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 10:18:28.659 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-14 10:18:29.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-14 10:18:29.658 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:2
08-14 10:18:30.160 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:3
08-14 10:18:30.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:4
08-14 10:18:31.158 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:5
08-14 10:18:31.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:6
08-14 10:18:32.158 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:7
08-14 10:18:32.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:8
08-14 10:18:33.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:9
08-14 10:18:33.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onComplete
mergeDelayError的作用和merge一样，它和merge的唯一区别是就算有一个或者多个Observable出现错误并不会马上停止合并结果，合适其它的结果都合并完成后再最后调用onErro方法。
mergeDelayError.png
例如沿用上面merge的方式改成mergeDelayError并补上一个erroObservable：
public static void mergeDelayErrorOperatorErro() {
//抛错误的Observable
Observable errorObservable = Observable.error(new Exception("this is end!")).delay(3500, TimeUnit.MILLISECONDS);
//被订阅者每隔1000毫秒后开始发射数据，并只取前面5个数据 发射顺序为0 2 4 6 8 第3500毫秒后产生错误
Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2;
return result;
}
}).take(5).mergeWith(errorObservable);
//被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2 + 1;
return result;
}
}).take(5);
Observable.mergeDelayError(observable1, observable2).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(XqTag.TAG, "onSubscribe");
}
@Override
public void onNext(Long value) {
Log.d(XqTag.TAG, "onNext:" + value);
}
@Override
public void onError(Throwable e) {
Log.d(XqTag.TAG, "onError:" + e.toString());
}
@Override
public void onComplete() {
Log.d(XqTag.TAG, "onComplete");
}
});
}
最后得到的结果是：
08-14 10:38:16.197 11015-11015/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 10:38:16.197 11015-11081/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-14 10:38:16.699 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-14 10:38:17.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:3
08-14 10:38:18.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:5
08-14 10:38:19.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:7
08-14 10:38:20.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:9
08-14 10:38:20.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onError:java.lang.Exception: this is end!
Zip操作符的作用是把n个observable提交的结果，严格按照顺序进行合并,最后这n个observable提交的后最终合并的结果的个数跟这n个observable里面发射个数最小那个Observable发射的个数一样。
20150713004523275.png
示例:
public static void zipOperator() {
//被订阅者每隔1000毫秒后开始发射数据，并只取前面3个数据 发射顺序为0 2 4
Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2;
return result;
}
}).take(3);
//被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
long result = aLong * 2 + 1;
return result;
}
}).take(5);
Observable.zip(observable1, observable2, new BiFunction<Long, Long, String>() {
@Override
public String apply(Long aLong, Long aLong2) throws Exception {
return aLong + "+" + aLong2 + "=" + (aLong + aLong2);
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(XqTag.TAG, "onSubscribe");
}
@Override
public void onNext(String value) {
Log.d(XqTag.TAG, "onNext:" + value);
}
@Override
public void onError(Throwable e) {
Log.d(XqTag.TAG, "onError:" + e.toString());
}
@Override
public void onComplete() {
Log.d(XqTag.TAG, "onComplete");
}
});
}
得到的结果是:
08-14 11:01:43.685 11940-11940/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 11:01:44.186 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:0+1=1
08-14 11:01:45.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:2+3=5
08-14 11:01:46.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:4+5=9
08-14 11:01:46.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onComplete
写在最后
写着写着已经深夜了。
最后附上源码:
https://git.oschina.net/jltx/RxJavaRetrofitDemoPro
