RxJava2 入门详细笔记
程序员文章站
2022-05-26 15:04:37
一、概述 在 RxJava 中，一个实现了 接口的对象可以订阅一个 类的实例。订阅者对 发射的任何数据或数据序列作出响应。这种模式简化了并发操作，因为它不需要阻塞等待 发射数据，而是创建了一个处于待命状态的观察者哨兵，哨兵在未来某个时刻响应 的通知。RxJava 提供了一套异步编程的 API，并且支 ......
一、概述
在 rxjava 中，一个实现了 observer 接口的对象可以订阅一个 observable 类的实例。订阅者对 observable 发射的任何数据或数据序列作出响应。这种模式简化了并发操作，因为它不需要阻塞等待 observable 发射数据，而是创建了一个处于待命状态的观察者哨兵，哨兵在未来某个时刻响应 observable 的通知。rxjava 提供了一套异步编程的 api，并且支持链式调用，所以使用 rxjava 编写的代码的逻辑会非常简洁
rxjava 有以下三个最基本的元素：
- 被观察者（observable）
- 观察者（observer）
- 订阅（subscribe）
创建被观察者
observable<integer> observable = observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> e) {
log.e(tag, "subscribe");
log.e(tag, "currentthread name: " + thread.currentthread().getname());
e.onnext(1);
e.onnext(2);
e.onnext(3);
e.oncomplete();
}
});
创建观察者
observer<integer> observer = new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext: " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
};
完成观察者与被观察者之间的订阅关系
observable.subscribe(observer);
也可以以链式调用的方式来完成订阅
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> e) {
log.e(tag, "subscribe");
log.e(tag, "currentthread name: " + thread.currentthread().getname());
e.onnext(1);
e.onnext(2);
e.onnext(3);
e.oncomplete();
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext: " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
最终的输出结果是一样的
onsubscribe
subscribe
currentthread name: main
onnext: 1
onnext: 2
onnext: 3
oncomplete
被观察者发送的事件类型有以下几种
| 事件种类 | 作用 |
|---|---|
| onnext() | 发送该事件时，观察者会回调 onnext() 方法 |
| onerror() | 发送该事件时，观察者会回调 onerror() 方法，当发送该事件之后，其他事件将不会继续发送 |
| oncomplete() | 发送该事件时，观察者会回调 oncomplete() 方法，当发送该事件之后，其他事件将不会继续发送 |
下面来讲解 rxjava 中各种常见的操作符
二、创建操作符
2.1、create()
用于创建一个 observable。一个正确的 observable 必须尝试调用观察者的 oncompleted 方法或者 onerror 方法有且仅有一次，而且此后不能再调用observable 的任何其它方法
observable<integer> observable = observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> e) {
log.e(tag, "subscribe");
log.e(tag, "currentthread name: " + thread.currentthread().getname());
e.onnext(1);
e.onnext(2);
e.onnext(3);
e.oncomplete();
}
});
2.2、just()
创建一个 observable并发送事件，发送的事件总数不可以超出十个
observable.just(1, 2, 3).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext: " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
onsubscribe
onnext: 1
onnext: 2
onnext: 3
oncomplete
2.3、fromarray
和 just() 类似，但 fromarray 可以传入多于十个的变量，并且可以传入一个数组
integer[] arrays = new integer[]{1, 2, 3};
observable.fromarray(arrays).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext: " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
2.4、fromcallable
这里的 callable 是指 java.util.concurrent 中的 callable，callable 和 runnable 的用法基本一致，只是它包含一个返回值，这个结果值就是发给观察者的
observable.fromcallable(new callable<integer>() {
@override
public integer call() {
return 100;
}
});
2.5、fromfuture
这里的 future 是指 java.util.concurrent 中的 future，future 的作用是增加了 cancel() 等方法操作 callable，它可以通过 get() 方法来获取 callable 返回的值
final futuretask<integer> futuretask = new futuretask<>(new callable<integer>() {
@override
public integer call() {
return 12;
}
});
observable.fromfuture(futuretask).doonsubscribe(new consumer<disposable>() {
@override
public void accept(disposable disposable) {
futuretask.run();
}
}).subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "accept: " + integer);
}
});
2.6、fromiterable()
用于发送一个 list 集合数据给观察者
list<integer> integerlist = new arraylist<>();
integerlist.add(1);
integerlist.add(2);
integerlist.add(3);
observable.fromiterable(integerlist).subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "accept: " + integer);
}
});
2.7、defer()
defer 操作符会一直等待直到有观察者订阅它，然后它使用 observable 工厂方法生成一个 observable。它对每个观察者都这样做，因此尽管每个订阅者都以为自己订阅的是同一个 observable ，实际上每个订阅者获取到的都是它们自己的单独的数据序列。在某些情况下，直到发生订阅时才生成 observable 可以确保 observable 包含最新的数据
//全局变量
private integer value = 100;
observable<integer> observable = observable.defer(new callable<observablesource<? extends integer>>() {
@override
public observablesource<? extends integer> call() {
return observable.just(value);
}
});
value = 200;
observable.subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "accept: " + integer);
}
});
value = 300;
observable.subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "accept: " + integer);
}
});
accept: 200
accept: 300
defer() 操作符能使得每次订阅操作都创建被观察者，因此两次订阅操作会创建不同的被观察者对象，因此两次打印操作返回的值并不一样
2.8、timer()
延迟指定时间后会发送一个大小为 0l 的值给观察者
observable.timer(2, timeunit.seconds)
.subscribe(new consumer<long>() {
@override
public void accept(long along) {
}
});
2.9、interval()
每隔一段时间就发送一个事件，传递的值从 0 开始并不断增 1
observable.interval(2, timeunit.seconds)
.subscribe(new consumer<long>() {
@override
public void accept(long along) {
log.e(tag, "value is: " + along);
}
});
2.10、intervalrange()
可以指定发送事件的开始值和数量，其他与 interval() 的功能一样
observable.intervalrange(2, 3, 4, 5, timeunit.seconds)
.subscribe(new observer<long>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(long along) {
log.e(tag, "onnext：" + along);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
起始值从 2 开始递增，事件共传递三次，第一次事件在订阅后延迟 4 秒触发，之后每次延迟 5 秒
10-06 10:48:40.017 17976-17976/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 10:48:44.017 17976-17990/leavesc.hello.rxjavademo e/mainactivity: onnext：2 10-06 10:48:49.017 17976-17990/leavesc.hello.rxjavademo e/mainactivity: onnext：3 10-06 10:48:54.017 17976-17990/leavesc.hello.rxjavademo e/mainactivity: onnext：4 10-06 10:48:54.017 17976-17990/leavesc.hello.rxjavademo e/mainactivity: oncomplete
2.11、range()
发送指定范围的事件序列
observable.range(2, 5)
.subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "values is :" + integer);
}
});
values is :2
values is :3
values is :4
values is :5
values is :6
2.12、rangelong()
作用与 range() 一样，只是数据类型是 long
observable.rangelong((2, 5)
.subscribe(new consumer<long>() {
@override
public void accept(long along) {
log.e(tag, "values is :" + along);
}
});
2.13、empty() & never() & error()
empty() 直接发送 oncomplete() 事件
observable.empty().subscribe(new observer<object>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(object object) {
log.e(tag, "onnext: " + object);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
打印结果
onsubscribe
oncomplete
换成 never()
onsubscribe
换成 error()
observable.error(new throwable("hello")).subscribe(new observer<object>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(object object) {
log.e(tag, "onnext: " + object);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
onsubscribe
onerror: hello
三、转换操作符
3.1、map()
map() 用于将被观察者发送的数据类型转变成其他的类型
observable.just(1, 2, 3)
.map(new function<integer, string>() {
@override
public string apply(integer integer) {
return "i'm " + integer;
}
})
.subscribe(new consumer<string>() {
@override
public void accept(string s) {
log.e(tag, s);
}
});
10-06 10:53:16.364 18099-18099/leavesc.hello.rxjavademo e/mainactivity: i'm 1 10-06 10:53:16.364 18099-18099/leavesc.hello.rxjavademo e/mainactivity: i'm 2 10-06 10:53:16.364 18099-18099/leavesc.hello.rxjavademo e/mainactivity: i'm 3
3.2、flatmap()
用于将事件序列中的元素进行整合加工，返回一个新的被观察者
list<list<string>> listarraylist = new arraylist<>();
list<string> stringlist = new arraylist<>();
for (int j = 0; j < 2; j++) {
stringlist.add("a_" + j);
}
listarraylist.add(stringlist);
stringlist = new arraylist<>();
for (int j = 0; j < 2; j++) {
stringlist.add("b_" + j);
}
listarraylist.add(stringlist);
observable.fromiterable(listarraylist).flatmap(new function<list<string>, observablesource<string>>() {
@override
public observablesource<string> apply(list<string> stringlist1) throws exception {
return observable.fromiterable(stringlist1);
}
}).subscribe(new consumer<string>() {
@override
public void accept(string s) throws exception {
log.e(tag, "value is: " + s);
}
});
10-06 11:02:47.246 18230-18230/leavesc.hello.rxjavademo e/mainactivity: value is: a_0 10-06 11:02:47.246 18230-18230/leavesc.hello.rxjavademo e/mainactivity: value is: a_1 10-06 11:02:47.246 18230-18230/leavesc.hello.rxjavademo e/mainactivity: value is: b_0 10-06 11:02:47.246 18230-18230/leavesc.hello.rxjavademo e/mainactivity: value is: b_1
3.3、concatmap()
concatmap() 和 flatmap() 基本一样，只不过 concatmap() 转发出来的事件是有序的，而 flatmap() 是无序的
还是用 flatmap()的例子来看
observable.fromiterable(listarraylist).flatmap(new function<list<string>, observablesource<string>>() {
@override
public observablesource<string> apply(list<string> stringlist1) throws exception {
if (stringlist1.get(0).startswith("a")) {
return observable.fromiterable(stringlist1).delay(200, timeunit.milliseconds);
}
return observable.fromiterable(stringlist1);
}
}).subscribe(new consumer<string>() {
@override
public void accept(string s) throws exception {
log.e(tag, "value is: " + s);
}
});
进行了一次延时操作，可以看到两次事件的发送顺序颠倒了
10-06 11:07:30.753 18702-18702/leavesc.hello.rxjavademo e/mainactivity: value is: b_0 10-06 11:07:30.753 18702-18702/leavesc.hello.rxjavademo e/mainactivity: value is: b_1 10-06 11:07:30.953 18702-18716/leavesc.hello.rxjavademo e/mainactivity: value is: a_0 10-06 11:07:30.953 18702-18716/leavesc.hello.rxjavademo e/mainactivity: value is: a_1
使用 concatmap() 则顺序将保持一致
3.4、buffer()
从需要发送的事件当中获取指定数量的事件，并将这些事件放到缓冲区当中一并发出。buffer 有两个参数，参数一count用于指点缓冲区大小，参数二 skip用指定当缓冲区满了时，发送下一次事件序列的时候要跳过多少元素
observable.just(1, 2, 3, 4, 5, 6)
.buffer(2, 2)
.subscribe(new observer<list<integer>>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(list<integer> integers) {
log.e(tag, "缓冲区大小： " + integers.size());
for (integer i : integers) {
log.e(tag, "元素： " + i);
}
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
onsubscribe 缓冲区大小： 2 元素： 1 元素： 2 缓冲区大小： 2 元素： 3 元素： 4 缓冲区大小： 2 元素： 5 元素： 6 oncomplete
3.5、groupby()
用于将数据进行分组，每个分组都会返回一个被观察者。groupby() 方法的返回值用于指定分组名，每返回一个新值就代表会创建一个分组
observable.just(1, 2, 3, 4, 5, 6, 7)
.groupby(new function<integer, string>() {
@override
public string apply(integer integer) {
if (integer < 4) {
return "hello";
}
return "hi";
}
})
.subscribe(new observer<groupedobservable<string, integer>>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(final groupedobservable<string, integer> observable) {
observable.subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "groupedobservable onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "groupedobservable onnext key :" + observable.getkey());
log.e(tag, "groupedobservable onnext value :" + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "groupedobservable onerror");
}
@override
public void oncomplete() {
log.e(tag, "groupedobservable oncomplete");
}
});
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror");
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 11:16:35.616 19015-19015/? e/mainactivity: onsubscribe 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onsubscribe 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hello 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :1 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hello 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :2 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hello 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :3 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onsubscribe 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hi 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :4 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hi 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :5 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hi 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :6 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext key :hi 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable onnext value :7 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable oncomplete 10-06 11:16:35.616 19015-19015/? e/mainactivity: groupedobservable oncomplete 10-06 11:16:35.616 19015-19015/? e/mainactivity: oncomplete
3.6、scan()
scan() 操作符对原始 observable 发射的第一条数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。它持续进行这个过程来产生剩余的数据序列
observable.just(1, 5, 8, 12).scan(new bifunction<integer, integer, integer>() {
@override
public integer apply(integer integer, integer integer2) {
log.e(tag, "integer : " + integer);
log.e(tag, "integer2 : " + integer2);
return integer + integer2;
}
}).subscribe(new consumer<integer>() {
@override
public void accept(integer integer) {
log.e(tag, "accept : " + integer);
}
});
10-06 11:25:19.389 19158-19158/leavesc.hello.rxjavademo e/mainactivity: accept : 1 10-06 11:25:19.389 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer : 1 10-06 11:25:19.389 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer2 : 5 10-06 11:25:19.399 19158-19158/leavesc.hello.rxjavademo e/mainactivity: accept : 6 10-06 11:25:19.399 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer : 6 10-06 11:25:19.399 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer2 : 8 10-06 11:25:19.399 19158-19158/leavesc.hello.rxjavademo e/mainactivity: accept : 14 10-06 11:25:19.399 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer : 14 10-06 11:25:19.409 19158-19158/leavesc.hello.rxjavademo e/mainactivity: integer2 : 12 10-06 11:25:19.409 19158-19158/leavesc.hello.rxjavademo e/mainactivity: accept : 26
四、组合操作符
4.1、concat() & concatarray()
用于将多个观察者组合在一起，然后按照参数的传入顺序发送事件，concat() 最多只可以发送4个事件
observable.concat(observable.just(1, 2),
observable.just(3, 4),
observable.just(5, 6),
observable.just(7, 8)).subscribe(new consumer<integer>() {
@override
public void accept(integer integer) throws exception {
log.e(tag, "accept: " + integer);
}
});
accept: 1 accept: 2 accept: 3 accept: 4 accept: 5 accept: 6 accept: 7 accept: 8
concatarray() 作用与 concat() 作用一样，不过前者可以发送多于 4 个的被观察者
4.2、merge() & mergearray()
这个方法与 concat() 作用基本一样，只是 concat() 是串行发送事件，而 merge() 并行发送事件
observable.merge(observable.interval(1, timeunit.seconds).map(new function<long, string>() {
@override
public string apply(long along) {
return "test_a_" + along;
}
}),
observable.interval(1, timeunit.seconds).map(new function<long, string>() {
@override
public string apply(long along) {
return "test_b_" + along;
}
})).subscribe(new consumer<string>() {
@override
public void accept(string s) {
log.e(tag, "accept: " + s);
}
});
test_a_0 test_b_0 test_a_1 test_b_1 test_a_2 test_b_2 test_b_3 test_a_3 test_a_4 test_b_4 test_a_5 test_b_5
mergearray() 可以发送 4 个以上的被观察者
4.3、concatarraydelayerror() & mergearraydelayerror()
在 concatarray() 和 mergearray() 两个方法当中，如果其中有一个被观察者发送了一个 error 事件，那么就会停止发送事件，如果想 onerror() 事件延迟到所有被观察者都发送完事件后再执行的话，可以使用 concatarraydelayerror() 和 mergearraydelayerror()
首先使用 concatarray() 来验证其发送 onerror() 事件是否会中断其他被观察者的发送事件
observable.concatarray(observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}), observable.just(30, 40, 50)).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext: " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror: " + e.getmessage());
}
@override
public void oncomplete() {
}
});
onnext: 1 onnext: 2 onerror: normal exception
从结果可以知道，确实中断了，现在换用 concatarraydelayerror()
10-06 04:00:04.935 6514-6514/? e/mainactivity: onnext: 1 10-06 04:00:04.935 6514-6514/? e/mainactivity: onnext: 2 10-06 04:00:04.935 6514-6514/? e/mainactivity: onnext: 30 10-06 04:00:04.935 6514-6514/? e/mainactivity: onnext: 40 10-06 04:00:04.935 6514-6514/? e/mainactivity: onnext: 50 10-06 04:00:04.935 6514-6514/? e/mainactivity: onerror: normal exception
从结果可以看到，onerror 事件是在所有被观察者发送完事件才发送的
4.4、zip()
zip() 操作符返回一个 obversable，它使用这个函数按顺序结合两个或多个 observables 发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个 observable 一样多的数据
observable.zip(observable.just(1, 2, 3, 4), observable.just(5, 6, 7, 8, 9),
new bifunction<integer, integer, string>() {
@override
public string apply(integer integer, integer integer2) throws exception {
return string.valueof(integer) + "_" + string.valueof(integer2);
}
})
.subscribe(new consumer<string>() {
@override
public void accept(string s) throws exception {
log.e(tag, "accept: " + s);
}
});
accept: 1_5 accept: 2_6 accept: 3_7 accept: 4_8
4.5、combinelatest() & combinelatestdelayerror()
combinelatest() 的作用与 zip() 类似，combinelatest() 发送事件的序列是与发送的时间线有关的，当两个 observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 observable 发射的最新数据，然后发射这个函数的结果
observable.zip(
observable.intervalrange(1, 4, 1, 1, timeunit.seconds)
.map(new function<long, string>() {
@override
public string apply(long along) {
string s1 = "a" + along;
log.e(tag, "a 发送的事件 " + s1);
return s1;
}
}), observable.intervalrange(1, 4, 2, 1, timeunit.seconds)
.map(new function<long, string>() {
@override
public string apply(long along) {
string s1 = "b" + along;
log.e(tag, "b 发送的事件 " + s1);
return s1;
}
}),
new bifunction<string, string, string>() {
@override
public string apply(string value1, string value2) throws exception {
return value1 + "_" + value2;
}
})
.subscribe(new consumer<string>() {
@override
public void accept(string s) throws exception {
log.e(tag, "accept: " + s);
}
});
10-06 05:17:06.337 7227-7241/leavesc.hello.rxjavademo e/mainactivity: a 发送的事件 a1 10-06 05:17:07.337 7227-7241/leavesc.hello.rxjavademo e/mainactivity: a 发送的事件 a2 10-06 05:17:07.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: b 发送的事件 b1 10-06 05:17:07.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: accept: a1_b1 10-06 05:17:08.337 7227-7241/leavesc.hello.rxjavademo e/mainactivity: a 发送的事件 a3 10-06 05:17:08.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: b 发送的事件 b2 10-06 05:17:08.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: accept: a2_b2 10-06 05:17:09.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: b 发送的事件 b3 10-06 05:17:09.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: accept: a3_b3 10-06 05:17:09.337 7227-7241/leavesc.hello.rxjavademo e/mainactivity: a 发送的事件 a4 10-06 05:17:10.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: b 发送的事件 b4 10-06 05:17:10.337 7227-7242/leavesc.hello.rxjavademo e/mainactivity: accept: a4_b4
当发送 a1 和 a2 事件时，b 并没有发送任何事件，所以不会触发到 accept 方法。当发送了 b1 事件之后，就会与 a 最新发送的事件 a2 结合成 a1_b2，之后的发射规则也以此类推
combinelatestdelayerror() 多了延迟发送 onerror() 的功能
4.6、reduce()
与 scan() 操作符的作用类似，也是将发送数据以一定逻辑聚合起来，区别在于 scan() 每处理一次数据就会将事件发送给观察者，而 reduce() 会将所有数据聚合在一起才会发送事件给观察者
observable.just(1, 3, 5, 7).reduce(new bifunction<integer, integer, integer>() {
@override
public integer apply(integer integer, integer integer2) throws exception {
log.e(tag, "integer1 : " + integer);
log.e(tag, "integer2 : " + integer2);
return integer + integer2;
}
}).subscribe(new consumer<integer>() {
@override
public void accept(integer integer) throws exception {
log.e(tag, "accept : " + integer);
}
});
integer1 : 1 integer2 : 3 integer1 : 4 integer2 : 5 integer1 : 9 integer2 : 7 accept : 16
4.7、collect()
collect() 与 reduce() 类似，但它的目的是收集原始 observable 发射的所有数据到一个可变的数据结构
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 {
log.e(tag, "accept : " + integers);
}
});
accept : [1, 2, 3, 4]
4.8、startwith() & startwitharray()
在发送事件之前追加事件，startwith() 追加一个事件，startwitharray() 可以追加多个事件，追加的事件会先发出
observable.just(4, 5)
.startwitharray(2, 3)
.startwith(1)
.subscribe(new consumer<integer>() {
@override
public void accept(integer integer) throws exception {
log.e(tag, "accept : " + integer);
}
});
10-06 05:38:21.081 8033-8033/leavesc.hello.rxjavademo e/mainactivity: accept : 1 10-06 05:38:21.081 8033-8033/leavesc.hello.rxjavademo e/mainactivity: accept : 2 10-06 05:38:21.081 8033-8033/leavesc.hello.rxjavademo e/mainactivity: accept : 3 10-06 05:38:21.081 8033-8033/leavesc.hello.rxjavademo e/mainactivity: accept : 4 10-06 05:38:21.081 8033-8033/leavesc.hello.rxjavademo e/mainactivity: accept : 5
4.9、count()
返回被观察者发送事件的数量
observable.just(1, 2, 3)
.count()
.subscribe(new consumer<long>() {
@override
public void accept(long along) throws exception {
log.e(tag, "along : " + along);
}
});
along : 3
五、功能操作符
5.1、delay()
延迟一段事件再发送事件
observable.just(1, 2, 3)
.delay(3, timeunit.seconds)
.subscribe(new consumer<integer>() {
@override
public void accept(integer value) throws exception {
log.e(tag, "value : " + value);
}
});
5.2、dooneach()
observable 发送一次事件之前都会回调这个方法
observable.just(1, 2, 3)
.dooneach(new consumer<notification<integer>>() {
@override
public void accept(notification<integer> integernotification) throws exception {
log.e(tag, "integernotification value : " + integernotification.getvalue());
}
})
.subscribe(new consumer<integer>() {
@override
public void accept(integer value) throws exception {
log.e(tag, "accept : " + value);
}
});
10-06 05:53:28.510 8645-8645/? e/mainactivity: integernotification value : 1 10-06 05:53:28.510 8645-8645/? e/mainactivity: accept : 1 10-06 05:53:28.510 8645-8645/? e/mainactivity: integernotification value : 2 10-06 05:53:28.510 8645-8645/? e/mainactivity: accept : 2 10-06 05:53:28.510 8645-8645/? e/mainactivity: integernotification value : 3 10-06 05:53:28.510 8645-8645/? e/mainactivity: accept : 3 10-06 05:53:28.510 8645-8645/? e/mainactivity: integernotification value : null
5.3、doonnext()
observable 发送 onnext() 之前都会先回调这个方法
observable.just(1, 2, 3)
.doonnext(new consumer<integer>() {
@override
public void accept(integer integer) throws exception {
log.e(tag, "doonnext accept : " + integer);
}
})
.subscribe(new consumer<integer>() {
@override
public void accept(integer value) throws exception {
log.e(tag, "accept : " + value);
}
});
10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 1 10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: accept : 1 10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 2 10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: accept : 2 10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 3 10-06 05:55:25.618 8758-8758/leavesc.hello.rxjavademo e/mainactivity: accept : 3
5.4、doafternext()
observable 发送 onnext() 之后都会回调这个方法
observable.just(1, 2, 3)
.doafternext(new consumer<integer>() {
@override
public void accept(integer integer) throws exception {
log.e(tag, "doonnext accept : " + integer);
}
})
.subscribe(new consumer<integer>() {
@override
public void accept(integer value) throws exception {
log.e(tag, "accept : " + value);
}
});
10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: accept : 1 10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 1 10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: accept : 2 10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 2 10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: accept : 3 10-06 05:57:09.357 8872-8872/leavesc.hello.rxjavademo e/mainactivity: doonnext accept : 3
5.5、dooncomplete()
observable 调用 oncomplete() 之前都会回调这个方法
observable.just(1, 2, 3)
.dooncomplete(new action() {
@override
public void run() throws exception {
log.e(tag, "dooncomplete run()");
}
})
.subscribe(new consumer<integer>() {
@override
public void accept(integer value) throws exception {
log.e(tag, "accept : " + value);
}
});
10-06 06:08:43.688 8982-8982/leavesc.hello.rxjavademo e/mainactivity: accept : 1 10-06 06:08:43.688 8982-8982/leavesc.hello.rxjavademo e/mainactivity: accept : 2 10-06 06:08:43.688 8982-8982/leavesc.hello.rxjavademo e/mainactivity: accept : 3 10-06 06:08:43.688 8982-8982/leavesc.hello.rxjavademo e/mainactivity: dooncomplete run()
5.6、doonerror()
observable 发送 onerror() 之前都会回调这个方法
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}).doonerror(new consumer<throwable>() {
@override
public void accept(throwable throwable) throws exception {
log.e(tag, "doonerror accept() : " + throwable.getmessage());
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
}
});
10-06 06:14:17.894 9230-9230/? e/mainactivity: onnext : 1 10-06 06:14:17.894 9230-9230/? e/mainactivity: onnext : 2 10-06 06:14:17.894 9230-9230/? e/mainactivity: doonerror accept() : normal exception 10-06 06:14:17.894 9230-9230/? e/mainactivity: onerror : normal exception
5.7、doonsubscribe()
observable 发送 onsubscribe() 之前会回调这个方法
5.8、doondispose()
当调用 disposable 的 dispose() 之后会回调该方法
5.9、doonlifecycle()
在回调 onsubscribe 之前回调该方法的第一个参数的回调方法，可以使用该回调方法决定是否取消订阅，doonlifecycle() 第二个参数的回调方法的作用与 doondispose() 一样
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.oncomplete();
}
}).doonlifecycle(new consumer<disposable>() {
@override
public void accept(disposable disposable) throws exception {
log.e(tag, "doonlifecycle accept");
}
}, new action() {
@override
public void run() throws exception {
log.e(tag, "doonlifecycle run");
}
}).subscribe(new observer<integer>() {
private disposable disposable;
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
this.disposable = d;
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
disposable.dispose();
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:31:45.011 9602-9602/leavesc.hello.rxjavademo e/mainactivity: doonlifecycle accept 10-06 06:31:45.011 9602-9602/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 06:31:45.011 9602-9602/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 06:31:45.011 9602-9602/leavesc.hello.rxjavademo e/mainactivity: doonlifecycle run
5.10、doonterminate() & doafterterminate()
doonterminate 是在 onerror 或者 oncomplete 发送之前回调，而 doafterterminate 则是 onerror 或者 oncomplete 发送之后回调
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.oncomplete();
}
}).doonterminate(new action() {
@override
public void run() throws exception {
log.e(tag, "doonterminate run");
}
}).doafterterminate(new action() {
@override
public void run() throws exception {
log.e(tag, "doafterterminate run");
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:34:55.968 9713-9713/? e/mainactivity: onsubscribe 10-06 06:34:55.968 9713-9713/? e/mainactivity: onnext : 1 10-06 06:34:55.968 9713-9713/? e/mainactivity: onnext : 2 10-06 06:34:55.968 9713-9713/? e/mainactivity: doonterminate run 10-06 06:34:55.968 9713-9713/? e/mainactivity: oncomplete 10-06 06:34:55.968 9713-9713/? e/mainactivity: doafterterminate run
5.11、dofinally()
在所有事件发送完毕之后回调该方法。 dofinally() 和 doafterterminate() 的区别在于取消订阅时，如果取消订阅，之后 doafterterminate() 就不会被回调，而 dofinally() 无论怎么样都会被回调，且都会在事件序列的最后
5.12、onerrorreturn()
当接受到一个 onerror() 事件之后回调，返回的值会回调 onnext() 方法，并正常结束该事件序列
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}).onerrorreturn(new function<throwable, integer>() {
@override
public integer apply(throwable throwable) throws exception {
return 7;
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:43:13.702 9946-9946/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 06:43:13.702 9946-9946/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 06:43:13.702 9946-9946/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 06:43:13.712 9946-9946/leavesc.hello.rxjavademo e/mainactivity: onnext : 7 10-06 06:43:13.712 9946-9946/leavesc.hello.rxjavademo e/mainactivity: oncomplete
5.13、onerrorresumenext()
当接收到 onerror() 事件时，返回一个新的 observable，并正常结束事件序列
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}).onerrorresumenext(new function<throwable, observablesource<? extends integer>>() {
@override
public observablesource<? extends integer> apply(throwable throwable) throws exception {
log.e(tag, "onerrorresumenext apply: " + throwable.getmessage());
return observable.just(4, 5, 6);
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onerrorresumenext apply: normal exception 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onnext : 4 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onnext : 5 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: onnext : 6 10-06 06:46:36.650 10243-10243/leavesc.hello.rxjavademo e/mainactivity: oncomplete
5.14、 onexceptionresumenext()
与 onerrorresumenext() 作用基本一致，但是这个方法只能捕捉 exception，不能捕获 error
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}).onexceptionresumenext(new observable<integer>() {
@override
protected void subscribeactual(observer<? super integer> observer) {
log.e(tag, "onexceptionresumenext subscribeactual");
observer.onnext(3);
observer.oncomplete();
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: onexceptionresumenext subscribeactual 10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: onnext : 3 10-06 06:51:49.396 10369-10369/leavesc.hello.rxjavademo e/mainactivity: oncomplete
将 emitter.onerror(new exception("normal exception")) 改为 emitter.onerror(new error("normal exception"));
异常将不会被捕获
10-06 06:53:21.655 10479-10479/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 06:53:21.655 10479-10479/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 06:53:21.655 10479-10479/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 06:53:21.655 10479-10479/leavesc.hello.rxjavademo e/mainactivity: onerror : normal exception
5.15、retry()
如果出现错误事件，则会重新发送所有事件序列指定次数
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new error("normal exception"));
}
}).retry(2).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 06:55:17.273 10591-10591/? e/mainactivity: onsubscribe 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 1 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 2 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 1 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 2 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 1 10-06 06:55:17.273 10591-10591/? e/mainactivity: onnext : 2 10-06 06:55:17.273 10591-10591/? e/mainactivity: onerror : normal exception
5.16、retryuntil()
出现错误事件之后，可以通过此方法判断是否继续发送事件
private int index = 1;
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.onerror(new exception("normal exception"));
}
}).retryuntil(new booleansupplier() {
@override
public boolean getasboolean() throws exception {
log.e(tag, "getasboolean");
return index == 7;
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
index++;
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: getasboolean 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: getasboolean 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 1 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onnext : 2 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: getasboolean 10-06 07:19:07.675 11433-11433/leavesc.hello.rxjavademo e/mainactivity: onerror : normal exception
5.17、repeat()
以指定次数重复发送被观察者的事件
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> emitter) throws exception {
emitter.onnext(1);
emitter.onnext(2);
emitter.oncomplete();
}
}).repeat(2).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
10-06 07:38:47.680 12155-12155/? e/mainactivity: onsubscribe 10-06 07:38:47.690 12155-12155/? e/mainactivity: onnext : 1 10-06 07:38:47.690 12155-12155/? e/mainactivity: onnext : 2 10-06 07:38:47.690 12155-12155/? e/mainactivity: onnext : 1 10-06 07:38:47.690 12155-12155/? e/mainactivity: onnext : 2 10-06 07:38:47.690 12155-12155/? e/mainactivity: oncomplete
5.18、repeatwhen()
返回一个新的被观察者来决定是否重复发送事件。如果新的被观察者返回 oncomplete 或者 onerror 事件，则旧的被观察者不会发送事件。如果新的被观察者返回其他事件，则旧的观察者会发送事件
observable.create(new observableonsubscribe<integer>() {
@override
public void subscribe(observableemitter<integer> e) throws exception {
e.onnext(1);
e.onnext(2);
e.onnext(3);
e.oncomplete();
}
}).repeatwhen(new function<observable<object>, observablesource<?>>() {
@override
public observablesource<?> apply(observable<object> objectobservable) throws exception {
// return observable.empty();
// return observable.error(new exception("normal exception"));
// return observable.just(1);
}
}).subscribe(new observer<integer>() {
@override
public void onsubscribe(disposable d) {
log.e(tag, "onsubscribe");
}
@override
public void onnext(integer integer) {
log.e(tag, "onnext : " + integer);
}
@override
public void onerror(throwable e) {
log.e(tag, "onerror : " + e.getmessage());
}
@override
public void oncomplete() {
log.e(tag, "oncomplete");
}
});
以上三种情况的输出结果分别是
10-06 14:29:05.641 20921-20921/leavesc.hello.rxjavademo e/mainactivity: onsubscribe 10-06 14:29:05.641 20921-20921/leavesc.hello.rxjavademo e/mainactivity: oncomplete
10-06 14:29:36.150 21027-21027/? e/mainactivity: onsubscribe 10-06 14:29:36.150 21027-21027/? e/mainactivity: onerror : normal exception
10-06 14:30:33.220 21135-21135/leavesc.hello.rxjavademo e/mainactivity: onsubscribe
10-06 14:30:33.220 21135-21135/leavesc.hello.rxjavademo e/mainactivity: onnext : 1
10-06 14:30:33.220 21135-21135/leavesc.hello.rxjavademo e/mainactivity: onnext : 2
10-06 14:30:33.220 21135-21135/leave
推荐阅读
-
ASP.NET MVC4入门教程（九）：查询详细信息和删除记录
-
超详细的Spring Boot入门笔记（总结）
-
ASP.NET MVC4入门教程（九）：查询详细信息和删除记录
-
笔记本电脑创建wifi热点共享的详细图文教程
-
有道云笔记APP怎么设置字体 有道云笔记设置字体的详细步骤
-
14寸纯白苹果本！海尔N1401简爱笔记本外观/详细参数评测
-
联想YOGA910笔记本开箱详细测评
-
联想yoga book怎么样？联想yoga book笔记本详细评测图解
-
戴尔Latitude 14 5000笔记本怎么样? 戴尔Latitude详细评测
-
小米笔记本Air和小新Air Pro哪个好？小米笔记本Air和联想小新Air Pro详细区别对比评测