# 简要：
需求了解：
Rxjava中有一些方便的辅助操作符，来更方便我们的函数式的编程。比如延迟、定时、指定操作的监听、数据类型转换等一系列的操作。
下面列出了一些用于Observable的辅助操作符：
Delay：延时发射Observable的结果。Do：注册一个动作作为原始Observable生命周期事件的监听器。SubscribeOn：指定Observable自身在哪个调度器上执行。ObserverOn：指定一个观察者在哪个调度器上观察这个Observable。Serialize：强制一个Observable连续调用并保证行为正确，其实就是同步事件操作。Materialize/Dematerialize：将数据项和事件通知都当做数据项发射 ，Dematerialize 刚好相反。TimeInterval：将一个发射数据的Observable转换为发射那些数据发射时间间隔的Observable。Timeout：对原始Observable的一个镜像，如果过了一个指定的时长仍没有发射数据，它会发一个错误通知。Timestamp：给Observable发射的数据项附加一个指定的时间戳。Using：创建一个只在Observable生命周期内存在的一次性资源。To：将Observable转换为另一个对象或数据结构。
# 1. Delay
延迟一段指定的时间再发射来自Observable的发射物。
Delay 操作符让原始 Observable 在发射每项数据之前都暂停一段指定的时间段。效果是Observable发射的数据项在时间上向前整体平移了一个增量。
1.1 delay(long delay, TimeUnit unit)
延迟指定时间段后发射原始Observable发射的数据序列，如果发生异常的话，会立即发射通知给观察者。
1.2 delay(Function<T, ObservableSource> itemDelay)
使用一个函数针对原始 Observable 的每一项数据返回一个 Observable ，它监视返回的这个 Observable ，当任何那样的 Observable 终止时，delay 返回的 Observable 就发射关联的那项数据。
1.3 delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay)
延迟直到 subscriptionDelay 发射第一个数据项后开始订阅原始 Observable，然后再使用一个函数针对原始Observable的每一项数据返回一个Observable，它监视返回的这个Observable，当任何那样的 Observable 终止时，delay 返回的 Observable 就发射关联的那项数据。
示例代码：
// 创建Observable
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
// emitter.onError(new Exception("Test Error!"));
emitter.onNext(4);
emitter.onNext(5);
emitter.onComplete();
}
});
/**
* 1. delay(long delay, TimeUnit unit,
* Scheduler scheduler: 可选参数，指定工作线程
* boolean delayError: 可选参数，延迟异常通知到最后
* )
* 延迟指定时间段后发射原始Observable发射的数据序列，如果发生异常的话，会立即发射通知给观察者。
*/
observable.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> doOnNext(1): " + integer);
}
}).delay(1, TimeUnit.SECONDS, Schedulers.newThread(), false) // 在子线程中延迟1秒发射数据，不延迟异常通知
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(1): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(1)");
}
});
System.in.read();
System.out.println("-----------------------------------------------------");
/**
* 2. delay(Function<T, ObservableSource<U>> itemDelay)
* 使用一个函数针对原始Observable的每一项数据返回一个Observable，它监视返回的这个Observable，
* 当任何那样的 Observable 终止时，delay 返回的 Observable 就发射关联的那项数据。
*/
observable.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> doOnNext(2): " + integer);
}
}).delay(new Function<Integer, ObservableSource<Long>>() {
@Override
public ObservableSource<Long> apply(Integer integer) throws Exception {
System.out.println("--> ObservableSource(2)： " + integer);
Observable<Long> timer = Observable.timer(integer, TimeUnit.SECONDS);
return timer;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(2): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(2)");
}
});
System.in.read();
System.out.println("-----------------------------------------------------");
/**
* 3. delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay)
* 延迟直到subscriptionDelay发射第一个数据项后开始订阅原始Observable
* 然后再使用一个函数针对原始Observable的每一项数据返回一个Observable，它监视返回的这个Observable，
* 当任何那样的 Observable 终止时，delay 返回的 Observable 就发射关联的那项数据。
*/
observable.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> doOnNext(3): " + integer);
}
// 延迟3秒后开始订阅源Observable，然后对发射的每项数据进行function函数延迟
}).delay(Observable.timer(3, TimeUnit.SECONDS), new Function<Integer, ObservableSource<Long>>() {
@Override
public ObservableSource<Long> apply(Integer integer) throws Exception {
System.out.println("--> apply(3): " + integer);
return Observable.timer(integer, TimeUnit.SECONDS);
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(3)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(3): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(3): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(3)");
}
});
System.in.read();
输出：
--> onSubscribe(1)
--> doOnNext(1): 1
--> doOnNext(1): 2
--> doOnNext(1): 3
--> doOnNext(1): 4
--> doOnNext(1): 5
--> onNext(1): 1
--> onNext(1): 2
--> onNext(1): 3
--> onNext(1): 4
--> onNext(1): 5
--> onCompleted(1)
-----------------------------------------------------
--> onSubscribe(2)
--> doOnNext(2): 1
--> ObservableSource(2)： 1
--> doOnNext(2): 2
--> ObservableSource(2)： 2
--> doOnNext(2): 3
--> ObservableSource(2)： 3
--> doOnNext(2): 4
--> ObservableSource(2)： 4
--> doOnNext(2): 5
--> ObservableSource(2)： 5
--> onNext(2): 1
--> onNext(2): 2
--> onNext(2): 3
--> onNext(2): 4
--> onNext(2): 5
--> onCompleted(2)
-----------------------------------------------------
--> onSubscribe(3)
--> doOnNext(3): 1
--> apply(3): 1
--> doOnNext(3): 2
--> apply(3): 2
--> doOnNext(3): 3
--> apply(3): 3
--> doOnNext(3): 4
--> apply(3): 4
--> doOnNext(3): 5
--> apply(3): 5
--> onNext(3): 1
--> onNext(3): 2
--> onNext(3): 3
--> onNext(3): 4
--> onNext(3): 5
--> onCompleted(3)
Javadoc: delay(long delay, TimeUnit unit, Scheduler scheduler, boolean delayError) (opens new window)
Javadoc: delay(Function<T, ObservableSource> itemDelay) (opens new window)
Javadoc: delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay) (opens new window)
# 2. Do
注册一个动作作为原始Observable生命周期事件的监听器。
你可以注册指定的回调，当Observable的某个事件发生时，Rxjava 会在与 Observable 链关联的正常通知集合中调用它。
在Rxjava中有许多相关Do的变体，分别进行不同场景的事件监听，一般有下面几种操作方法：
doOnSubscribe(Consumer onSubscribe)：一旦有观察者订阅了Observable，就会被调用。doOnLifecycle(Consumer onSubscribe, Action onDispose)： 在观察者订阅产生和解除时被调用。doOnNext(Consumer onNext)：在 Observable 每次发射数据前被调用。doOnEach(Observer observer): 在 Observable 调用观察者的所有通知前被调用。doAfterNext(Consumer onAfterNext)：在 Observable 调用OnNext通知(数据发射通知)之后被调用。doOnError(Consumer onError)：注册一个动作，当它的 Observable 由于异常终止调用 onError 时会被调用。doOnTerminate(Action onTerminate): 当Observable终止之前会被调用，无论是正常还是异常终止。doAfterTerminate(Action onFinally): 当Observable终止之后会被调用，无论是正常还是异常终止。doOnComplete(Action onComplete)：Observable正常终止调用 onCompleted 时会被调用。doFinally(Action onFinally)：Observable终止之后会被调用，无论是正常还是异常终止，但是优先于doAfterTerminate。doOnDispose(Action onDispose)：在观察者调用Disposable的dispose()方法时被调用。
示例代码：
/**
* 1. doOnSubscribe(Consumer onSubscribe)
* 一旦有观察者订阅了Observable，就会被调用
*/
Observable.just(999).doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(Disposable disposable) throws Exception {
System.out.println("----> doOnSubscribe");
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(1): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(1)");
}
});
System.out.println("--------------------------------------------");
/**
* 2. doOnLifecycle(Consumer onSubscribe, Action onDispose)
* onSubscribe： 接受观察者订阅前的通知，可以在此通知中解除订阅
* onDispose： 接受观察者调用解除订阅通知
* 在观察者订阅产生和解除时调用
*/
Observable.just(999).doOnLifecycle(new Consumer<Disposable>() {
@Override
public void accept(Disposable disposable) throws Exception {
System.out.println("----> doOnLifecycle onSubscribe(2)");
// disposable.dispose(); // 可以在观察者订阅前直接解除订阅
}
}, new Action() {
@Override
public void run() throws Exception {
System.out.println("----> doOnLifecycle onDispose(2)");
}
}).subscribe(new Observer<Integer>() {
private Disposable disposable;
@Override
public void onSubscribe(Disposable d) {
disposable = d;
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(2): " + integer);
disposable.dispose(); // 手动解除订阅
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(2)");
}
});
System.out.println("--------------------------------------------");
/**
* 3. doOnNext(Consumer onNext)
* 在Observable每次发射数据前被调用
*/
Observable.just(999).doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("----> doOnNext(3): " + integer);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> accept(3): " + integer);
}
});
System.out.println("--------------------------------------------");
/**
* 4. doOnEach(Observer observer)
* 在Observable调用观察者的所有通知前被调用
*/
Observable.just(999).doOnEach(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("----> doOnEach(4) onSubscribe");
}
@Override
public void onNext(Integer integer) {
System.out.println("----> doOnEach(4) onNext: " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("----> doOnEach(4) onError: " + e);
}
@Override
public void onComplete() {
System.out.println("----> doOnEach(4) onComplete");
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(4)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(4): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(4): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onCompleted(4)");
}
});
System.out.println("--------------------------------------------");
/**
* 5. doAfterNext(Consumer onAfterNext)
* 在Observable调用OnNext通知(数据发射通知)之后被调用
*/
Observable.just(999).doAfterNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("----> doAfterNext(5): " + integer);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> onNext(5): " + integer);
}
});
System.out.println("--------------------------------------------");
/**
* 6. doOnError(Consumer onError)
* 注册一个动作，当它的Observable由于异常终止调用 onError 时会被调用
*/
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onError(new Exception("Test Error!"));
}
}).doOnError(new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
System.out.println("----> doOnError(6): " + throwable);
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(6)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(6): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(6): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(6)");
}
});
System.out.println("--------------------------------------------");
/**
* 7.
* doOnTerminate(Action onTerminate): 当Observable终止之前会被调用，无论是正常还是异常终止
* doAfterTerminate(Action onFinally): 当Observable终止之后会被调用，无论是正常还是异常终止
*/
Observable.just(999).doOnTerminate(new Action() {
@Override
public void run() throws Exception {
System.out.println("----> doOnTerminate(7)");
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> accept(7): " + integer);
}
});
System.out.println("--------------------------------------------");
/**
* 8. doOnComplete(Action onComplete)
* Observable正常终止调用 onCompleted 时会被调用
*/
Observable.just(999).doOnComplete(new Action() {
@Override
public void run() throws Exception {
System.out.println("----> doOnComplete(8)");
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(8)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(8): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(8): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(8)");
}
});
System.out.println("--------------------------------------------");
/**
* 9. doFinally(Action onFinally)
* Observable终止之后会被调用，无论是正常还是异常终止，但是优先于doAfterTerminate
*/
Observable.just(999).doFinally(new Action() {
@Override
public void run() throws Exception {
System.out.println("----> doFinally(9)");
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(9)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(9): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(9): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(9)");
}
});
System.out.println("--------------------------------------------");
/**
* 10. doOnDispose(Action onDispose)
* 在观察者调用Disposable的dispose()方法时被调用
*/
Observable.just(999).doOnDispose(new Action() {
@Override
public void run() throws Exception {
System.out.println("----> doOnDispose(10)");
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(10)");
d.dispose();
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(10): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(10): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(10)");
}
});
输出：
----> doOnSubscribe
--> onSubscribe(1)
--> onNext(1): 999
--> onCompleted(1)
--------------------------------------------
----> doOnLifecycle onSubscribe(2)
--> onSubscribe(2)
--> onNext(2): 999
----> doOnLifecycle onDispose(2)
--------------------------------------------
----> doOnNext(3): 999
--> accept(3): 999
--------------------------------------------
--> onSubscribe(4)
----> doOnEach(4) onNext: 999
--> onNext(4): 999
----> doOnEach(4) onComplete
--> onCompleted(4)
--------------------------------------------
--> onNext(5): 999
----> doAfterNext(5): 999
--------------------------------------------
--> onSubscribe(6)
----> doOnError(6): java.lang.Exception: Test Error!
--> onError(6): java.lang.Exception: Test Error!
--------------------------------------------
--> accept(7): 999
----> doOnTerminate(7)
--------------------------------------------
--> onSubscribe(8)
--> onNext(8): 999
----> doOnComplete(8)
--> onComplete(8)
--------------------------------------------
--> onSubscribe(9)
--> onNext(9): 999
--> onComplete(9)
----> doFinally(9)
--------------------------------------------
--> onSubscribe(10)
----> doOnDispose(10)
Javadoc: doOnSubscribe(Consumer onSubscribe) (opens new window)
Javadoc: doOnLifecycle(Consumer onSubscribe, Action onDispose) (opens new window)
Javadoc: doOnNext(Consumer onNext) (opens new window)
Javadoc: doOnEach(Observer observer) (opens new window)
Javadoc: doAfterNext(Consumer onAfterNext) (opens new window)
Javadoc: doOnError(Consumer onError) (opens new window)
Javadoc: doOnTerminate(Action onTerminate) (opens new window)
Javadoc: doAfterTerminate(Action onFinally) (opens new window)
Javadoc: doOnComplete(Action onComplete) (opens new window)
Javadoc: doFinally(Action onFinally) (opens new window)
Javadoc: doOnDispose(Action onDispose) (opens new window)
# 3. SubscribeOn
指定Observable自身在哪个调度器上执行。
使用调度器 Scheduler 来管理多线程环境中Observable的转场。你可以使用 SubscribeOn 操作符指定Observable在一个特定的调度器上运转。
示例代码：
// 查看当前线程id
System.out.println("----> main: threadID = " + Thread.currentThread().getId());
/**
* subscribeOn(Scheduler scheduler)
* 指定Observable在指定的scheduler上调度
*/
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 查看Observable的工作线程id
System.out.println("----> SubscribeOn: threadID = " + Thread.currentThread().getId());
emitter.onNext(999);
emitter.onComplete();
}
}).subscribeOn(Schedulers.newThread()) // 指定Observable的工作线程在子线程
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("--> accept: " + integer);
}
});
System.in.read();
输出：
----> main: threadID = 1
----> SubscribeOn: threadID = 13
--> accept: 999
Javadoc: subscribeOn(Scheduler scheduler) (opens new window)
# 4. ObserverOn
指定一个观察者在哪个调度器上观察这个Observable。
使用调度器 Scheduler 来管理多线程环境中Observable的转场。你可以使用 ObserveOn 操作符指定Observable在一个特定的调度器上发送通知给观察者 (调用观察者的onNext, onCompleted , onError 方法)。
示例代码：
// 查看当前线程id
System.out.println("----> main: threadID = " + Thread.currentThread().getId());
/**
* observeOn(Scheduler scheduler,
* boolean delayError, // 可选参数是否延迟异常
* int bufferSize // 指定缓存大小
* )
* 指定观察者在指定的scheduler线程中调度
*/
Observable.just(999).observeOn(Schedulers.newThread(), true, 3)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
// 查看观察者的线程id
System.out.println("--> accept ThreadID: " + Thread.currentThread().getId());
System.out.println("--> accept: " + integer);
}
});
System.in.read();
输出：
----> main: threadID = 1
--> accept ThreadID: 13
--> accept: 999
Javadoc: observeOn(Scheduler scheduler) (opens new window)
Javadoc: observeOn(Scheduler scheduler, boolean delayError) (opens new window)
Javadoc: observeOn(Scheduler scheduler, boolean delayError, int bufferSize) (opens new window)
# 5. Serialize
强制一个Observable连续调用并保证行为正确，其实就是同步事件操作。
一个Observable可以异步调用它的观察者的方法，可能是从不同的线程调用。这可能会让Observable行为不正确，它可能会在某一个 onNext 调用之前尝试调用 onCompleted 或 onError 方法，或者从两个不同的线程同时调用 onNext 方法。使用 Serialize 操作符，你可以纠正这个Observable的行为，保证它的行为是正确的且是同步的。
示例代码：
/**
* serialize()
* 强制一个Observable连续调用(同步)并保证行为正确
*/
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 多线程事件调用
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
emitter.onNext(i + 1);
}
emitter.onComplete();
}
}).start();
// 多线程事件调用
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 100; i < 110; i++) {
emitter.onNext(i + 1);
}
emitter.onComplete();
}
}).start();
}
}).serialize() // 序列化,合法性操作
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext: " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});
System.in.read();
输出：
---------------------------------------------
下面是没有使用Serialize()场景，发现不合法的调用
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onNext: 5
--> onNext: 6
--> onNext: 7
--> onNext: 8
--> onNext: 101
--> onNext: 102
--> onNext: 103
--> onNext: 104
--> onNext: 105
--> onNext: 9
--> onNext: 106
--> onNext: 10
--> onNext: 107
--> onComplete
--> onNext: 108 // 不合法的调用
----------------------------------------------
下面是使用Serialize()场景，合法的事件调用
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onNext: 5
--> onNext: 6
--> onNext: 7
--> onNext: 8
--> onNext: 9
--> onNext: 101
--> onNext: 102
--> onNext: 103
--> onNext: 104
--> onNext: 105
--> onNext: 106
--> onNext: 107
--> onNext: 108
--> onNext: 109
--> onNext: 110
--> onComplete
Javadoc: serialize() (opens new window)
# 6. Materialize
Materialize 将数据项和事件通知都当做数据项发射。
一个合法的有限的Obversable将调用它的观察者的 onNext 方法零次或多次，然后调用观察者的 onCompleted 或 onError 正好一次。 Materialize 操作符将这一系列调用，包括原来的 onNext 通知和终止通知onCompleted 或 onError 都转换为一个Observable发射的数据序列。
解析： 将来自原始Observable的通知转换为 Notification 对象，然后它返回的Observable会发射这些数据。
示例代码：
/**
* materialize()
* 将来自原始Observable的通知转换为Notification对象，然后它返回的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("Test Error!"));
emitter.onComplete();
}
}).materialize()
.subscribe(new Observer<Notification<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}
@Override
public void onNext(Notification<Integer> integerNotification) {
System.out.println("--> onNext: " + integerNotification);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});
输出：
--> onSubscribe
--> onNext: OnNextNotification[1]
--> onNext: OnNextNotification[2]
--> onNext: OnErrorNotification[java.lang.Exception: Test Error!]
--> onComplete
Javadoc: materialize() (opens new window)
# 7. Dematerialize
Dematerialize 操作符是 Materialize 的逆向过程，它将 Materialize 转换的结果还原成它原本的形式。
解析： dematerialize 反转这个过程，将原始Observable发射的 Notification 对象还原成Observable的通知。
示例代码：
/**
* dematerialize()
* 过时的方法，在Rxjava：2.2.4中已经被dematerialize(Function<T, Notification<R>> selector)替代
* 将原始Observable发射的 Notification 对象还原成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("Test Error!"));
emitter.onComplete();
}
}).materialize()
.dematerialize() // 将Notification 对象还原成Observable的通知
.subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Object o) {
System.out.println("--> onNext(1): " + o);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(1)");
}
});
System.out.println("------------------------------------------------");
/**
* dematerialize(Function<T, Notification<R>> selector)
* 将原始Observable发射的 Notification 对象经过一个selector函数处理后，发射一个新的Notification，
* 还原成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("Test Error!"));
emitter.onComplete();
}
}).materialize()
.dematerialize(new Function<Notification<Integer>, Notification<Integer>>() {
@Override
public Notification<Integer> apply(Notification<Integer> integerNotification) throws Exception {
System.out.println("--> apply(2): " + integerNotification);
return integerNotification;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Integer integer) {
System.out.println("--> onNext(2): " + integer);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(2)");
}
});
输出：
--> onSubscribe(1)
--> onNext(1): 1
--> onNext(1): 2
--> onError(1): java.lang.Exception: Test Error!
------------------------------------------------
--> onSubscribe(2)
--> apply(2): OnNextNotification[1]
--> onNext(2): 1
--> apply(2): OnNextNotification[2]
--> onNext(2): 2
--> apply(2): OnErrorNotification[java.lang.Exception: Test Error!]
--> onError(2): java.lang.Exception: Test Error!
Javadoc: dematerialize() (opens new window)
Javadoc: dematerialize(Function<T,Notification<R>>selector) (opens new window)
# 8. TimeInterval
将一个发射数据的Observable转换为发射那些数据发射时间间隔的Observable。
TimeInterval 操作符拦截原始Observable发射的数据项，替换为发射表示相邻发射物时间间隔的对象。
这个操作符将原始 Observable 转换为另一个 Observable ，后者发射一个标志替换前者的数据项，这个标志表示前者的两个连续发射物之间流逝的时间长度。新的Observable的第一个发射物表示的是在观察者订阅原始 Observable 到原始 Observable 发射它的第一项数据之间流逝的时间长度。不存在与原始 Observable 发射最后一项数据和发射 onCompleted 通知之间时长对应的发射物。
示例代码：
/**
* 1. timeInterval(Scheduler scheduler)
* scheduler: 可选参数，指定调度线程
* 接收原始数据项，发射射表示相邻发射物时间间隔的对象
*/
Observable.intervalRange(1, 10, 100, 100, TimeUnit.MILLISECONDS)
.timeInterval()
// .timeInterval(Schedulers.newThread()) // 指定工作线程
.subscribe(new Observer<Timed<Long>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Timed<Long> longTimed) {
long time = longTimed.time(); // 连续数据间的间隔时间
TimeUnit unit = longTimed.unit(); // 连续数据间的时间间隔单位
Long value = longTimed.value(); // Observable发送的数据项
System.out.println("--> onNext(1): " + longTimed.toString());
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(1)");
}
});
System.in.read();
System.out.println("-------------------------------------------------");
/**
* 2. timeInterval(TimeUnit unit, Scheduler scheduler)
* 指定时间间隔单位和指定工作线程，接收原始数据项，发射射表示相邻发射物时间间隔的对象
*/
Observable.intervalRange(1, 10, 1000, 1200, TimeUnit.MILLISECONDS)
// .timeInterval(TimeUnit.SECONDS) // 指定时间间隔单位
.timeInterval(TimeUnit.SECONDS, Schedulers.newThread()) // 指定时间间隔单位和指定工作线程
.subscribe(new Observer<Timed<Long>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Timed<Long> longTimed) {
System.out.println("--> onNext(2): " + longTimed.toString());
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(2)");
}
});
System.in.read();
输出：
--> onSubscribe(1)
--> onNext(1): Timed[time=104, unit=MILLISECONDS, value=1]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=2]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=3]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=4]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=5]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=6]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=7]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=8]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=9]
--> onNext(1): Timed[time=100, unit=MILLISECONDS, value=10]
--> onComplete(1)
-------------------------------------------------
--> onSubscribe(2)
--> onNext(2): Timed[time=1, unit=SECONDS, value=1]
--> onNext(2): Timed[time=1, unit=SECONDS, value=2]
--> onNext(2): Timed[time=1, unit=SECONDS, value=3]
--> onNext(2): Timed[time=1, unit=SECONDS, value=4]
--> onNext(2): Timed[time=1, unit=SECONDS, value=5]
--> onNext(2): Timed[time=2, unit=SECONDS, value=6]
--> onNext(2): Timed[time=1, unit=SECONDS, value=7]
--> onNext(2): Timed[time=1, unit=SECONDS, value=8]
--> onNext(2): Timed[time=1, unit=SECONDS, value=9]
--> onNext(2): Timed[time=1, unit=SECONDS, value=10]
--> onComplete(2)
Javadoc: timeInterval() (opens new window)
Javadoc: timeInterval(Scheduler scheduler) (opens new window)
Javadoc: timeInterval(TimeUnit unit) (opens new window)
Javadoc: timeInterval(TimeUnit unit, Scheduler scheduler) (opens new window)
# 9. Timeout
对原始Observable的一个镜像，如果过了一个指定的时长仍没有发射数据，它会发一个错误通知。
RxJava中的实现为 timeout 操作符，具有多个不同的变体。
# 9.1 timeout(timeout, timeUnit)
如果原始 Observable 过了指定的一段时长没有发射任何数据，Timeout操作符会以一个 onError 通知终止这个Observable。
示例代码：
/**
* 1. timeout(long timeout, TimeUnit timeUnit)
* 接受一个时长参数，如果在指定的时间段内没有数据项发射，将会发射一个Error通知，
* 或者每当原始Observable发射了一项数据， timeout 就启动一个计时器，
* 如果计时器超过了指定指定的时长而原始Observable没有发射另一项数据，
* 就抛出 TimeoutException ，以一个错误通知终止Observable。
*/
Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
// Thread.sleep(2000); // 延迟2秒后发射数据，此时会有TimeoutException
emitter.onNext(1L);
Thread.sleep(2000); // 延迟2秒后发射数据，此时会有TimeoutException
emitter.onNext(2L);
emitter.onComplete();
}
}).timeout(1, TimeUnit.SECONDS) // 指定超时时间段为1秒
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Long aLong) {
System.out.println("--> onNext(1): " + aLong);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(1)");
}
});
System.in.read();
输出：
--> onSubscribe(1)
--> onNext(1): 1
--> onError(1): java.util.concurrent.TimeoutException: The source did not signal an event for 1 seconds and has been terminated.
Javadoc: timeout(long timeout, TimeUnit timeUnit) (opens new window)
# 9.2 timeout(timeout, timeUnit, scheduler, other)
在指定时间段后超时时会切换到使用一个你指定的备用的 Observable，而不是发onError通知，可以通过scheduler 来指定工作线程。
示例代码：
/**
* 2. timeout(long timeout, TimeUnit timeUnit,
* Scheduler scheduler, // 可选参数，指定线程调度器
* ObservableSource other // 可选参数，超时备用Observable
* )
*
* 在指定时间段后超时时会切换到使用一个你指定的备用的Observable，而不是发onError通知。
*/
Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
// Thread.sleep(2000); // 延迟2秒后发射数据，此时会有TimeoutException
emitter.onNext(1L);
Thread.sleep(2000); // 延迟2秒后发射数据，此时会有TimeoutException
emitter.onNext(2L);
emitter.onComplete();
}
}).timeout(1, TimeUnit.SECONDS, // 指定超时时间段为1秒
Schedulers.newThread(), // 指定工作线程为子线程
Observable.just(888L)) // 超时后默认发射的Observable
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Long aLong) {
System.out.println("--> onNext(2): " + aLong);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(2)");
}
});
System.in.read();
输出：
--> onSubscribe(2)
--> onNext(2): 1
--> onNext(2): 888
--> onComplete(2)
# 9.3 timeout(Function itemTimeoutIndicator, ObservableSource other)
使用一个函数 itemTimeoutIndicator 针对原始 Observable 的每一项返回一个 Observable，如果当这个 Observable 终止时原始 Observable 还没有发射另一项数据，就会认为是超时了，如果没有指定超时备用的 other，就抛出 TimeoutException，以一个错误通知终止 bservable，否则超时后发射备用的 Observable。
示例代码：
/**
* 3. timeout(Function<T, ObservableSource> itemTimeoutIndicator
* ObservableSource other // 可选参数，当超时后发射的备用Observable
* )
* 对原始Observable的每一项返回一个Observable，
* 如果当这个Observable终止时原始Observable还没有发射另一项数据，就会认为是超时了，
* 如果没有指定超时备用的Observable，就抛出TimeoutException，以一个错误通知终止Observable，
* 否则超时后发射备用的Observable。
*/
Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(ObservableEmitter<Long> emitter) throws Exception {
emitter.onNext(1L);
Thread.sleep(3000); // 延迟3秒后发射数据，此时会有TimeoutException
emitter.onNext(2L);
emitter.onComplete();
}
}).timeout(new Function<Long, ObservableSource<Long>>() {
@Override
public ObservableSource<Long> apply(Long aLong) throws Exception {
// 为每一个原始数据发射一个Observable来指示下一个数据发射的Timeout，这里指定1秒超时时间
return Observable.timer(1, TimeUnit.SECONDS);
}
}, Observable.just(888L)) // 超时后默认发射的Observable
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(3)");
}
@Override
public void onNext(Long aLong) {
System.out.println("--> onNext(3): " + aLong);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(3): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(3)");
}
});
System.in.read();
输出：
--> onSubscribe(3)
--> onNext(3): 1
--> onNext(3): 888
--> onComplete(3)
Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator) (opens new window)
Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator, ObservableSource other) (opens new window)
# 10. Timestamp
给Observable发射的数据项附加一个指定的时间戳。
timestamp ，它将一个发射Timed类型数据的Observable转换为一个发射类型为 Timestamped<Timed> 的数据的Observable，每一项都包含数据的原始发射时间信息和原始数据。
示例代码：
/**
* 1. timestamp(Scheduler scheduler)
* scheduler: 可选参数，指定线程调度器
*
* 给Observable发射的数据项附加一个时间戳信息
*/
Observable.intervalRange(1, 5, 1, 100, TimeUnit.MILLISECONDS)
.timestamp(Schedulers.newThread()) // 指定在子线程调度处理
.subscribe(new Observer<Timed<Long>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(1)");
}
@Override
public void onNext(Timed<Long> longTimed) {
long time = longTimed.time(); // 连续数据间的间隔时间
TimeUnit unit = longTimed.unit(); // 连续数据间的时间间隔单位
Long value = longTimed.value(); // Observable发送的数据项
System.out.println("--> onNext(1): " + longTimed);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(1): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(1)");
}
});
System.in.read();
System.out.println("-------------------------------------------");
/**
* 2. timestamp(TimeUnit unit, Scheduler scheduler)
* scheduler: 可选参数，指定线程调度器
*
* 给Observable发射的数据项附加一个指定单位的时间戳信息
*/
Observable.intervalRange(1, 5, 1, 1200, TimeUnit.MILLISECONDS)
.timestamp(TimeUnit.SECONDS, Schedulers.newThread()) // 指定时间单位为秒，在子线程调度处理
.subscribe(new Observer<Timed<Long>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onNext(Timed<Long> longTimed) {
System.out.println("--> onNext(2): " + longTimed);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete(2)");
}
});
System.in.read();
输出：
--> onSubscribe(1)
--> onNext(1): Timed[time=1577455367446, unit=MILLISECONDS, value=1]
--> onNext(1): Timed[time=1577455367545, unit=MILLISECONDS, value=2]
--> onNext(1): Timed[time=1577455367645, unit=MILLISECONDS, value=3]
--> onNext(1): Timed[time=1577455367745, unit=MILLISECONDS, value=4]
--> onNext(1): Timed[time=1577455367845, unit=MILLISECONDS, value=5]
--> onComplete(1)
-------------------------------------------
--> onSubscribe(2)
--> onNext(2): Timed[time=1577455369, unit=SECONDS, value=1]
--> onNext(2): Timed[time=1577455370, unit=SECONDS, value=2]
--> onNext(2): Timed[time=1577455371, unit=SECONDS, value=3]
--> onNext(2): Timed[time=1577455373, unit=SECONDS, value=4]
--> onNext(2): Timed[time=1577455374, unit=SECONDS, value=5]
--> onComplete(2)
Javadoc: timestamp() (opens new window)
Javadoc: timestamp(Scheduler scheduler) (opens new window)
Javadoc: timestamp(TimeUnit unit) (opens new window)
Javadoc: timestamp(TimeUnit unit, Scheduler scheduler) (opens new window)
# 11. Using
创建一个只在Observable生命周期内存在的一次性资源。
Using 操作符让你可以指示Observable创建一个只在它的生命周期内存在的资源，当Observable终止时这个资源会被自动释放。
using 操作符接受三个参数：
observableFactory：一个用户创建一次性资源的工厂函数resourceFactory：一个用于创建Observable的工厂函数disposeFunction：一个用于释放资源的函数
当一个观察者订阅 using 返回的Observable时， using 将会使用Observable工厂函数创建观察者要观察的Observable，同时使用资源工厂函数创建一个你想要创建的资源。当观察者取消订阅这个Observable时，或者当观察者终止时（无论是正常终止还是因错误而终止）， using 使用第三个函数释放它创建的资源。
示例代码：
/**
* 用于在Observable的生命周期内存在的资源对象
*/
class MyResource {
private String resource;
public MyResource(String resource) {
this.resource = resource;
}
@Override
public String toString() {
return "MyResource{" +
"resource='" + resource + '\'' +
'}';
}
public void releaseResource() {
System.out.println("----> MyResource resource is release. ");
resource = null;
}
}
/**
* 1. using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer, boolean eager)
*
* resourceSupplier: // 一个用户创建一次性资源的工厂函数
* sourceSupplier: // 一个用于创建Observable的工厂函数
* disposer: // 一个用于释放资源的函数
* eager: // 可选参数，如果为true的话，则第三个函数disposer的处理在Observable的结束前执行
*
* 当一个观察者订阅 using 返回的Observable时， using 将会使用Observable工厂函数创建观察者要观察的Observable，
* 同时使用资源工厂函数创建一个你想要创建的资源。
* 当观察者取消订阅这个Observable时，或者当观察者终止时（无论是正常终止还是因错误而终止），
* using 使用第三个函数释放它创建的资源。
*/
Observable.using(
// 一个用户创建一次性资源的工厂函数
new Callable<MyResource>() {
@Override
public MyResource call() throws Exception {
System.out.println("----> resourceSupplier call");
return new MyResource("This is Observable resource!");
}
},
// 一个用于创建Observable的工厂函数，这个函数返回的Observable就是最终被观察的Observable
new Function<MyResource, ObservableSource<Long>>() {
@Override
public ObservableSource<Long> apply(MyResource myResource) throws Exception {
System.out.println("----> sourceSupplier apply: " + myResource);
return Observable.rangeLong(1, 5);
}
},
// 一个用于释放资源的函数
new Consumer<MyResource>() {
@Override
public void accept(MyResource myResource) throws Exception {
System.out.println("----> disposer accept: ");
myResource.releaseResource();
}
},
// 可选参数，如果为true的话，则在Observable的结束前执行释放资源的函数
true).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}
@Override
public void onNext(Long aLong) {
System.out.println("--> onNext: " + aLong);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}
@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});
输出：
----> resourceSupplier call(1)
----> sourceSupplier apply(1): MyResource{resource='This is Observable resource!'}
--> onSubscribe(1)
--> onNext(1): 1
--> onNext(1): 2
--> onNext(1): 3
--> onNext(1): 4
--> onNext(1): 5
----> disposer accept(1):
----> MyResource resource is release.
--> onComplete
Javadoc: using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer) (opens new window)
Javadoc: using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer, boolean eager) (opens new window)
# 12. To
将Observable转换为另一个对象或数据结构。
将 Observable 或者Observable 发射的数据序列转换为另一个对象或数据结构。它们中的一些会阻塞直到 Observable 终止，然后生成一个等价的对象或数据结构；另一些返回一个发射那个对象或数据结构的 Observable。
由于 rxjava 的 To 操作符中有很多 toXXX 操作符的实现和不同的变体重载，此处就不详细的展开了，有兴趣的可以查看官方的API 文档 (opens new window) 详细参阅。
下面几个是常见的几种To操作符的：
toList()：让Observable将多项数据组合成一个List，然后调用一次onNext方法传递整个列表。toMap(Function keySelector,Function valueSelector)：toMap收集原始Observable发射的所有数据项到一个Map（默认是HashMap）然后发射这个Map。 你可以提供一个用于生成Map的Key的函数，还可以提供一个函数转换数据项到Map存储的值（默认数据项本身就是值）。toSortedList()： 它会对产生的列表排序，默认是自然升序，如果发射的数据项没有实现Comparable接口，会抛出一个异常，你也可以传递一个函数作为用于比较两个数据项。toMultimap(Function keySelector, Function valueSelector)：类似于toMap，不同的是，它生成的这个Map的value类型还是一个ArrayList。
示例代码：
/**
* 1. toList()
* 让Observable将多项数据组合成一个List，然后调用一次onNext方法传递整个列表。
*/
range.toList()
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(List<Integer> integers) throws Exception {
System.out.println("--> toList accept(1): " + integers);
}
});
System.out.println("------------------------------------------");
/**
* 2. toMap(Function<? super T, ? extends K> keySelector,Function<? super T, ? extends V> valueSelector)
* toMap收集原始Observable发射的所有数据项到一个Map（默认是HashMap）然后发射这个Map。
* 你可以提供一个用于生成Map的Key的函数，还可以提供一个函数转换数据项到Map存储的值（默认数据项本身就是值）。
*/
range.toMap(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "key" + integer; // 返回一个Map的key
}
}, new Function<Integer, Integer>() {
@Override
public Integer apply(Integer integer) throws Exception {
return integer; // 返回一个Map的value
}
}).subscribe(new SingleObserver<Map<String, Integer>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(2)");
}
@Override
public void onSuccess(Map<String, Integer> stringIntegerMap) {
System.out.println("--> onSuccess(2): " + stringIntegerMap);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(2): " + e);
}
});
System.out.println("------------------------------------------");
/**
* 3. toSortedList()
* 它会对产生的列表排序，默认是自然升序，如果发射的数据项没有实现Comparable接口，会抛出一个异常。
* 然而，你也可以传递一个函数作为用于比较两个数据项
*/
Observable.just(5, 3, 8, 6, 9, 10)
.toSortedList()
.subscribe(new SingleObserver<List<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(3)");
}
@Override
public void onSuccess(List<Integer> integers) {
System.out.println("--> onSuccess(3): " + integers);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(3): " + e);
}
});
System.out.println("------------------------------------------");
/**
* 4. toSortedList(Comparator comparator)
*
* 传递一个函数comparator作为用于比较两个数据项，它会对产生的列表排序
*/
Observable.just(5, 3, 8, 6, 9, 10)
.toSortedList(new Comparator<Integer>() {
@Override
public int compare(Integer o1, Integer o2) {
System.out.println("--> compare: o1 = " + o1 + ", o2 = " + o2);
return o1 - o2; // 比较器的排序逻辑
}
}).subscribe(new SingleObserver<List<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(4)");
}
@Override
public void onSuccess(List<Integer> integers) {
System.out.println("--> onSuccess(4): " + integers);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(4): " + e);
}
});
System.out.println("------------------------------------------");
/**
* 5. toMultimap(Function<T, K> keySelector, Function<T, V> valueSelector)
* 类似于 toMap ，不同的是，它生成的这个Map的value类型还是一个ArrayList
*/
range.toMultimap(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "key" + integer; // 返回一个Map的key
}
}, new Function<Integer, Integer>() {
@Override
public Integer apply(Integer integer) throws Exception {
return integer; // 返回一个Map的value
}
}).subscribe(new SingleObserver<Map<String, Collection<Integer>>>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe(5)");
}
@Override
public void onSuccess(Map<String, Collection<Integer>> stringCollectionMap) {
System.out.println("--> onSuccess(5): " + stringCollectionMap);
}
@Override
public void onError(Throwable e) {
System.out.println("--> onError(5): " + e);
}
});
输出：
--> toList accept(1): [1, 2, 3, 4, 5]
------------------------------------------
--> onSubscribe(2)
--> onSuccess(2): {key1=1, key2=2, key5=5, key3=3, key4=4}
------------------------------------------
--> onSubscribe(3)
--> onSuccess(3): [3, 5, 6, 8, 9, 10]
------------------------------------------
--> onSubscribe(4)
--> compare: o1 = 3, o2 = 5
--> compare: o1 = 8, o2 = 3
--> compare: o1 = 8, o2 = 5
--> compare: o1 = 6, o2 = 5
--> compare: o1 = 6, o2 = 8
--> compare: o1 = 9, o2 = 6
--> compare: o1 = 9, o2 = 8
--> compare: o1 = 10, o2 = 6
--> compare: o1 = 10, o2 = 9
--> onSuccess(4): [3, 5, 6, 8, 9, 10]
------------------------------------------
--> onSubscribe(5)
--> onSuccess(5): {key1=[1], key2=[2], key5=[5], key3=[3], key4=[4]}
Javadoc: toList() (opens new window)
Javadoc: toMap(Function keySelector,Function valueSelector) (opens new window)
Javadoc: toSortedList() (opens new window)
Javadoc: toMultimap(Function keySelector, Function valueSelector) (opens new window)
# 小结
本节主要是介绍了 Rxjava 中的各种辅助操作符，比如延迟、超时，事件监听等相关的辅助类型的操作，这在开发中是很有用处的。
提示：以上使用的Rxjava2版本: 2.2.12 (opens new window)
实例代码：