目录html
接续上篇: Rxjava2 Observable的辅助操做详解及实例(一)java
将一个发射数据的Observable转换为发射那些数据发射时间间隔的Observable。react
TimeInterval
操做符拦截原始Observable发射的数据项,替换为发射表示相邻发射物时间间隔的对象。git
这个操做符将原始 Observable 转换为另外一个 Observable ,后者发射一个标志替换前者的数据项,这个标志表示前者的两个连续发射物之间流逝的时间长度。新的Observable的第一个发射物表示的是在观察者订阅原始 Observable 到原始 Observable 发射它的第一项数据之间流逝的时间长度。不存在与原始 Observable 发射最后一项数据和发射 onCompleted 通知之间时长对应的发射物。github
示例代码:数据结构
/** * 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();
输出:app
--> 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()
Javadoc: timeInterval(Scheduler scheduler)
Javadoc: timeInterval(TimeUnit unit)
Javadoc: timeInterval(TimeUnit unit, Scheduler scheduler)ide
对原始Observable的一个镜像,若是过了一个指定的时长仍没有发射数据,它会发一个错误通知。函数
RxJava中的实现为 timeout
操做符,具备多个不一样的变体。this
若是原始 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.
在指定时间段后超时时会切换到使用一个你指定的备用的 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)
Javadoc: timeout(long timeout, TimeUnit timeUnit, Scheduler scheduler, 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)
Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator, ObservableSource other)
给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()
Javadoc: timestamp(Scheduler scheduler)
Javadoc: timestamp(TimeUnit unit)
Javadoc: timestamp(TimeUnit unit, Scheduler scheduler)
建立一个只在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)
Javadoc: using(Callable resourceSupplier, Function sourceSupplier, Consumer disposer, boolean eager)
将Observable转换为另外一个对象或数据结构。
将 Observable 或者Observable 发射的数据序列转换为另外一个对象或数据结构。它们中的一些会阻塞直到 Observable 终止,而后生成一个等价的对象或数据结构;另外一些返回一个发射那个对象或数据结构的 Observable。
因为 rxjava 的 To
操做符中有不少 toXXX
操做符的实现和不一样的变体重载,此处就不详细的展开了,有兴趣的能够查看官方的API 文档 详细参阅。
下面几个是常见的几种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()
Javadoc: toMap(Function keySelector,Function valueSelector)
Javadoc: toSortedList()
Javadoc: toMultimap(Function keySelector, Function valueSelector)
本节主要是介绍了 Rxjava
中的各类辅助操做符,好比延迟、超时,事件监听等相关的辅助类型的操做,这在开发中是颇有用处的。
提示:以上使用的Rxjava2版本: 2.2.12
Rx介绍与讲解及完整目录参考:Rxjava2 介绍与详解实例
实例代码: