版权声明:转载前请留言获得作者许可,转载后标明作者 张拭心 与 原文链接。大家都是成年人,创作不易,感谢您的支持! https://blog.csdn.net/u011240877/article/details/75282790
本篇文章是阅读 官方文档 的笔记。
作者:shixinzhang(百度搜索 “shixinzhang CSDN” 即可找到我)
RxJava 也用了有段时间,那么多操作符总不想去记,遇到了才去查一下,查的次数多了感觉效率太低,还不如先过一遍官方文档。
因此接下来几篇我们一起刷一遍 RxJava 官方文档,这个过程可能会很枯燥,但是像电影里少林寺练功 一样,只有先通过枯燥的学习掌握基本功,然后才能考虑如何应用,加油!
![这里写图片描述](https://img-blog.csdn.net/20170712081931407?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
读完本文你将了解:
创建型操作符
创建型操作符即用于创建 Observable 的操作符。
create
![这里写图片描述](https://img-blog.csdn.net/20170711191858250?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中,create()
通过一个传递一个 Observable.OnSubscribe
创建一个Observable
,Observable.OnSubscribe
的泛型参数指明要创建的数据类型:
mObservable = Observable.create(new Observable.OnSubscribe<String>() {
/**
* 当 observable 被订阅时,会自动调用 call() 方法,依次触发其中的事件
* 其实就是调用订阅者的回调方法,即实现了被观察者向观察者的事件传递
* @param subscriber
*/
@Override
public void call(Subscriber<? super String> subscriber) {
if (!subscriber.isUnsubscribed()) {
for (int i = 0; i < 5; i ++) {
subscriber.onNext("next 1");
}
subscriber.onCompleted();
}
subscriber.onError(new Throwable("Error message"));
}
});
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
在 create()
中根据创建任务的状态分别调用 onNext()
, onCompleted()
和 onError()
。
在 create()
中最好调用 isUnsubscribed()
检查观察者的订阅状态,这样在没有观察者时可以避免做无用的创建工作。
create()
默认不在任何特定的调度器上执行。
defer
![这里写图片描述](https://img-blog.csdn.net/20170711191927368?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
defer
中文意思 “推迟、延迟”。
defer
操作符,只有观察者订阅后才会使用一个 Observable
工厂方法创建 Observable
,每次有新的观察者订阅时,都会重复这个操作。
订阅者以为订阅的是同一个数据源,其实是各自订阅的是不同的 Observable
。
RxJava 中对应的实现是 defer()
,接受一个 Func0<Observable<T>>
类型的参数:
public final static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
return create(new OnSubscribeDefer<T>(observableFactory));
}
我们可以传入一个 Func0<Observable<T>>
的匿名内部类,泛型参数是数据类型:
mObservable = Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
return "test";
}
});
Empty
Empty
用于创建一个不发射任何数据,但是会发射终止消息的 Observable
。
![这里写图片描述](https://img-blog.csdn.net/20170711191958684?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中对应的实现是 empty()
方法:
public final static <T> Observable<T> empty() {
return (Observable<T>) EmptyHolder.INSTANCE;
}
private static final class EmptyHolder {
final static Observable<Object> INSTANCE = create(new OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
subscriber.onCompleted();
}
});
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
使用起来也很简单,没有参数:
mObservable = Observable.empty();
Never
Never
用于创建一个不发射任何数据,也不会发射终止消息的 Observable
(意义何在?)。
![这里写图片描述](https://img-blog.csdn.net/20170711192028574?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中对应的实现是 never()
方法:
public final static <T> Observable<T> never() {
return NeverObservable.instance();
}
private static class NeverObservable<T> extends Observable<T> {
private static class Holder {
static final NeverObservable<?> INSTANCE = new NeverObservable<Object>();
}
@SuppressWarnings("unchecked")
static <T> NeverObservable<T> instance() {
return (NeverObservable<T>) Holder.INSTANCE;
}
NeverObservable() {
super(new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> observer) {
}
});
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
可以看到这个方法返回的 NeverObservable
不会发出任何消息,使用起来和 empty()
类似:
//不会发出任何消息
mObservable = Observable.never();
Throw
Throw
用于创建一个不发射任何数据,但是会发射错误消息的 Observable
。
![这里写图片描述](https://img-blog.csdn.net/20170711192101727?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中对应的实现是 error()
方法:
public final static <T> Observable<T> error(Throwable exception) {
return new ThrowObservable<T>(exception);
}
private static class ThrowObservable<T> extends Observable<T> {
public ThrowObservable(final Throwable exception) {
super(new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> observer) {
observer.onError(exception);
}
});
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
可以看到,error()
会发射一个 onError()
事件。
使用很简单,只需传入一个 Throwable
:
mObservable = Observable.error(new Throwable("error message"));
Empty/Never/Throw
三个操作符创建的 Observable 比较特殊,一般用于测试,有时也结合其它的数据源,作为其它操作符的参数。
From
From
可以将其他对象或者数据类型转换成 Observables
。
![这里写图片描述](https://img-blog.csdn.net/20170712081701595?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
当使用 Observable 时,最好是所有数据源都是 Observable 类型,那样配合操作符将十分方便。这时如果有 Observable 以外的其他数据类型,我们最好使用一些操作符将其他类型转换为 Observable 。
比如 Iterable
迭代器,可以转换成一系列同步的 Observable;Future
,可以转做每次只发射一个元素的 Observable。通过显式地将其他类型转换成 Observable,就可以方便地体会 Rx 的便捷之处。
大部分 ReactiveX 的实现语言都提供了将特定的对象和数据结构转换为 Observables 的方法。
RxJava 中对应的实现是 from()
:
public final static <T> Observable<T> from(T[] array) {
return from(Arrays.asList(array));
}
public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
return create(new OnSubscribeFromIterable<T>(iterable));
}
public static <T> Observable<T> from(Future<? extends T> future) {
return (Observable<T>)unsafeCreate(OnSubscribeToObservableFuture.toObservableFuture(future));
}
from()
支持参数为数组、Iterable、Future,也就是说,这个操作符可以将数组、Iterable 和 Future 类型的数据转换为 Observable。
对于 Iterable 和数组,转换后的 Observable 会发射 Iterable 或数组中的每一项数据。
不了解 Iterable 的同学可以看这篇:Java 集合源码解析:Iterator。
使用例子:
private void createObservables() {
String[] words = {"shixin", "is", "cute"};
Observable<String> from = Observable.from(words);
from.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
System.out.println("onNext: " + s);
}
});
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170712081718668?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
from()
默认不在任何特定的调度器上执行。我们可以将 Scheduler
作为第二个参数传递给Observable,这个 Future 将会在指定的线程执行。
Interval
Interval
用于创建一个按指定时间间隔、发送递增的整数序列的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170712081727232?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
Interval
操作符返回的 Observable 会间隔发射一个无限递增的整数序列。
RxJava 中对应的实现为 interval()
,它接受一个表示时间间隔的参数和一个表示时间单位的参数:
public static Observable<Long> interval(long interval, TimeUnit unit) {
return interval(interval, interval, unit, Schedulers.computation());
}
public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
return unsafeCreate(new OnSubscribeTimerPeriodically(initialDelay, period, unit, scheduler));
}
使用例子:
private void createObservableWithInterval() {
Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);
interval.subscribe(getPrintSubscriber());
}
/**
* 用于打印结果的订阅者
* @param <T>
* @return
*/
private <T> Subscriber<T> getPrintSubscriber() {
return new Subscriber<T>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("onError: " + e.getMessage());
}
@Override
public void onNext(T t) {
System.out.println("onNext: " + t);
if ((Long)t > 10){
unsubscribe();
}
}
};
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
在这个例子中我们使用 interval()
创建了一个 Observable,它将发射递增的整数序列。在 onNext()
中我们当 t > 10 时取消订阅。运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170712081739823?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
Just
Just
用于创建一个发射特定元素的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170712081752345?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
Just
操作符用于将一个元素转换成 Observable,然后将它发射出去。
Just
和 From
有些相似,不同之处在于 From
会将数组或 Iterable
的数据取出后逐个发射;而 Just
只是简单的原样发射,将数组或 Iterable 当做单个数据一次性发射出去。
RxJava 中对应的实现是 just()
:
public static <T> Observable<T> just(final T value) {
return ScalarSynchronousObservable.create(value);
}
public static <T> Observable<T> just(T t1, T t2) {
return from((T[])new Object[] { t1, t2 });
}
可以看到,多个参数的 just()
最后其实还是调用的 from()
。
使用例子:
private void createObservableWithJust() {
String[] words = {"shixin", "is", "cute"};
Observable<String[]> just = Observable.just(words);
just.subscribe(getPrintSubscriber());
}
我们传递给 just()
一个数组,结果输出的还是数组对象;传递给一个 null,输出的也是 null:
![这里写图片描述](https://img-blog.csdn.net/20170712081802823?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
可以看到,just()
果然和名字一样,就是将转换的数据直接发出去。
注意,如果你传递 null 给 just,它会返回一个 Observable 然后发射出去一个 null,而不会创建一个空的 Observable。如果想创建一个空的,你需要使用 Empty
操作符。
Range
Range
用于创建一个发射指定范围的整数序列的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170712081811823?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中对应的实现为 range()
:
public static Observable<Integer> range(int start, int count) {
if (count < 0) {
throw new IllegalArgumentException("Count can not be negative");
}
if (count == 0) {
return Observable.empty();
}
if (start > Integer.MAX_VALUE - count + 1) {
throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
}
if (count == 1) {
return Observable.just(start);
}
return Observable.unsafeCreate(new OnSubscribeRange(start, start + (count - 1)));
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
可以看到,just()
接受两个参数,第一个是起始值,第二个是个数。如果你将第二个值设置为 0,就不会发射数据;如果设置为负数或者起始值太大,就会抛出异常。
使用例子:
private void createObservableWithRange() {
Observable<Integer> range = Observable.range(3, 5);
range.subscribe(this.<Integer>getPrintSubscriber());
}
运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170712081823771?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
Repeat
Repeat
用于创建一个多次发射特定数据的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170711192252138?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中的实现 repeat()
不是静态的,也就是说它不可以用于创建 Observable,只可以对已有的 Observable 进行重复发射,参数指定重复次数。
public final Observable<T> repeat(final long count) {
return OnSubscribeRedo.<T>repeat(this, count);
}
使用时需要先创建一个 Observable,然后对他进行 repeat 操作:
private void createObservableWithRepeat() {
String[] words = {"shixin", "is", "cute"};
Observable<String> from = Observable.from(words);
from.repeat(2)
.subscribe(getPrintSubscriber());
}
运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170712081834189?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
可以看到,当 repeat()
接收到 onCompleted()
事件后触发重订阅。
repeat
操作符默认在 trampoline
调度器上执行。我们可以自行指定 Scheduler。
RepeatWhen
RepeatWhen 相对费解一些。
Repeat
接收到 onCompleted()
事件后就会触发重订阅,而 RepeatWhen
则在它的基础上,增加了什么时候重订阅的控制。
![这里写图片描述](https://img-blog.csdn.net/20170711192333374?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RepeatWhen
在接受到 onCompleted 事件后,会进行条件检查,然后进行相应的重发操作;在接收到 onError 事件后,会停止重复。
RxJava 中对应的实现是 repeatWhen()
:
public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler) {
return OnSubscribeRedo.repeat(this, InternalObservableUtils.createRepeatDematerializer(notificationHandler));
}
可以看到, repeatWhen()
的第一个参数是输入的 Observable 的处理函数,第二个参数是处理策略。
由于只有在调用 onCompleted 时才会调用参数,所以它是 Void 类型。
使用例子:
Observable.unsafeCreate(new Observable.OnSubscribe<String>() {
@Override
public void call(final Subscriber<? super String> subscriber) {
for (int i = 0; i < 5; i++) {
subscriber.onNext("item " + i);
}
subscriber.onCompleted();
}
}).repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
@Override
public Observable<?> call(final Observable<? extends Void> completed) {
return completed.delay(5, TimeUnit.SECONDS);
}
}).subscribe(getPrintSubscriber());
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
每次发射 onCompleted 事件时会进入 repeatWhen
的匿名内部类参数,在其中我们做了延迟处理。
运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170711192412669?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
可以看到,每过 5 秒重复一次。
Start
Start
的作用是创建一个发射函数返回值的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170711192432095?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
编程语言中有很多种获取计算结果的方法,比如 functions
, futures
, actions
, callables
, runnables
等等。Start
操作符可以将他们的结果转为 Observable,从而更方便地进行操作。
RxJava 中 Start
的实现有很多种,它们都属于 rxjava-async
模块。
在 RxJava 中 From
操作符可以将 Future
转换为 Observable,与 start
相似。
Timer
Timer
用于创建一个在指定延迟后发射值的 Observable。
![这里写图片描述](https://img-blog.csdn.net/20170711192449289?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
RxJava 中对应的实现是 timer
:
public static Observable<Long> timer(long delay, TimeUnit unit) {
return timer(delay, unit, Schedulers.computation());
}
public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
return unsafeCreate(new OnSubscribeTimerOnce(delay, unit, scheduler));
}
timer()
返回一个Observable,它会在延迟给定的时间后发射一个数字 0。
private void createObservableWithTimer() {
Observable.timer(3, TimeUnit.SECONDS)
.subscribe(this.<Long>getPrintSubscriber());
}
运行结果:
![这里写图片描述](https://img-blog.csdn.net/20170711192503183?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvdTAxMTI0MDg3Nw==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
我们走完了约五分之一的流程,坚持就是胜利!!!
Thanks
http://reactivex.io/documentation/operators.html
http://www.jianshu.com/p/023a5f60e6d0