Delay —延迟一段指定的时间再发送来自Observable的项目
private static void delay() {
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
System.out.println("emit time: " + System.currentTimeMillis());
for (int i = 0; i < 5; i++) {
e.onNext(i);
}
}
})
.delay(1, TimeUnit.SECONDS)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("onNext: " + integer + "time: " + System.currentTimeMillis());
}
});
while (true) ;
}
输出结果:
emit time: 1509262337562
onNext: 0time: 1509262338568
onNext: 1time: 1509262338568
onNext: 2time: 1509262338568
onNext: 3time: 1509262338568
onNext: 4time: 1509262338568
Do — 注册一个动作作为原始Observable生命周期事件的一种占位符
主要有doAfterTerminate、doOnComplete、doOnDispose、doOnEach doOnError、doOnLifecycle、doOnNext、doOnSubscribe、 doOnTerminate、onTerminateDetach
private static void doOnNext() {
Observable
.just(1, 2, 3)
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("doOnNext " + integer);
}
})
.doAfterTerminate(new Action() {
@Override
public void run() throws Exception {
System.out.println("doAfterTerminate");
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("onNext " + integer);
}
});
}
输出结果:
doOnNext 1
onNext 1
doOnNext 2
onNext 2
doOnNext 3
onNext 3
doAfterTerminate
Materialize/Dematerialize — Materialize将数据项和事件通知都当做数据项发射，Dematerialize刚好相反。Dematerialize操作符是Materialize的逆向过程，它将Materialize转换的结果还原成它原本的形式。
ObserveOn — 指定Observable在一个特定的调度器上发送通知给观察者 (调用观察者的onNext, onCompleted, onError方法)
private static void observeOn() {
Observable
.just(1, 2, 3)
.observeOn(Schedulers.newThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println(Thread.currentThread() + " onNext " + integer);
}
});
while (true);
}
输出结果：
Thread[RxNewThreadScheduler-1,5,main] onNext 1
Thread[RxNewThreadScheduler-1,5,main] onNext 2
Thread[RxNewThreadScheduler-1,5,main] onNext 3
Serialize — Observable可以异步调用它的观察者的方法，可能是从不同的线程调用。这可能会让Observable行为不正确，它可能会在某一个onNext调用之前尝试调用onCompleted或onError方法，或者从两个不同的线程同时调用onNext方法。使用Serialize操作符，你可以纠正这个Observable的行为，保证它的行为是正确的且是同步的。
Subscribe — 操作来自Observable的发射物和通知
SubscribeOn — 指定Observable在一个特定的调度器上运转
private static void subscribeOn() {
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
for (int i = 0; i < 3; i++) {
e.onNext(Thread.currentThread() + " - " + i);
}
}
})
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(Thread.currentThread() + " onNext " + s);
}
});
while (true) ;
}
输出结果：
Thread[RxNewThreadScheduler-1,5,main] onNext Thread[RxCachedThreadScheduler-1,5,main] - 0
Thread[RxNewThreadScheduler-1,5,main] onNext Thread[RxCachedThreadScheduler-1,5,main] - 1
Thread[RxNewThreadScheduler-1,5,main] onNext Thread[RxCachedThreadScheduler-1,5,main] - 2
TimeInterval — 操作符拦截原始Observable发射的数据项，转换为发射表示相邻发射物时间间隔的对象。
private static void timeInterval() {
Observable
.just(1, 2, 3, 4, 5, 6)
.timeInterval()
.subscribe(new SafeObserver<Timed<Integer>>(new Observer<Timed<Integer>>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(@NonNull Timed<Integer> integerTimed) {
System.out.println(integerTimed.time() + " onNext " + integerTimed.value());
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
}));
}
输出结果：
onSubscribe
1 onNext 1
1 onNext 2
0 onNext 3
0 onNext 4
0 onNext 5
0 onNext 6
onComplete
Timeout — 对原始Observable的一个镜像，如果过了一个指定的时长仍没有发射数据，它会发一个错误通知
private static void timeout() {
Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; i < 5; i++) {
Thread.sleep(i * 200);
e.onNext(i);
}
}
})
.timeout(500, TimeUnit.MILLISECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
System.out.println("onNext " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError " + e.toString());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
while (true) ;
}
输出结果：
onSubscribe
onNext 0
onNext 1
onNext 2
onError java.util.concurrent.TimeoutException
Timestamp —给Observable发送的项目加上一个时间戳
private static void timeStamp() {
Observable
.just(1, 2, 3, 4, 5)
.timestamp(TimeUnit.MILLISECONDS)
.subscribe(new Consumer<Timed<Integer>>() {
@Override
public void accept(Timed<Integer> i) throws Exception {
System.out.println(i.value() + " - " + i.time());
}
});
}
输出结果：
1 - 1509456831220
2 - 1509456831221
3 - 1509456831221
4 - 1509456831221
5 - 1509456831221
Using — 创建与Observable具有相同寿命的一次性资源
using操作符接受三个参数：
- 一个创建一次性资源的
- 一个创建Observable
- 一个释放资源的函数
private static void using() {
Observable
.using(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("one ");
return 5;
}
}, new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
System.out.println("two ");
return Observable.just("emit " + integer);
}
}, new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("three ");
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println("onNext " + s);
}
});
}
输出结果：
one
two
onNext emit 5
three
To — 将Observable转换为另一个对象或数据结构
具体有：blockingIterable blockingLatest blockingMostRecent blockingNext sorted to toFuture toList toMap toMultimap toSortedList
private static void to() {
Observable
.just(1, 2, 3, 4, 5)
.toList()
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(List<Integer> integers) throws Exception {
for (Integer integer : integers) {
System.out.println(integer);
}
}
});
}
输出结果：
1
2
3
4
5
