Rxjava2 Observable的辅助操作详解及实例

Posted 2020-11-25 jiangming-blogs

目录

• 8. TimeInterval
• 9. Timeout
  • 9.1 timeout(timeout, timeUnit)
  • 9.2 timeout(timeout, timeUnit, scheduler, other)
  • 9.3 timeout(Function itemTimeoutIndicator, ObservableSource other)
• 10. Timestamp
• 11. Using
• 12. To
• 小结

接续上篇： Rxjava2 Observable的辅助操作详解及实例（一）

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()
Javadoc: timeInterval(Scheduler scheduler)
Javadoc: timeInterval(TimeUnit unit)
Javadoc: timeInterval(TimeUnit unit, Scheduler scheduler)

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)

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)

Javadoc: timeout(long timeout, TimeUnit timeUnit, Scheduler scheduler, ObservableSource other)

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)
Javadoc: timeout(Function<T, ObservableSource> itemTimeoutIndicator, ObservableSource other)

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()
Javadoc: timestamp(Scheduler scheduler)
Javadoc: timestamp(TimeUnit unit)
Javadoc: timestamp(TimeUnit unit, Scheduler scheduler)

11. Using

创建一个只在Observable生命周期内存在的一次性资源。

Using 操作符让你可以指示Observable创建一个只在它的生命周期内存在的资源，当Observable终止时这个资源会被自动释放。

using 操作符接受三个参数：

1. observableFactory：一个用户创建一次性资源的工厂函数
2. resourceFactory：一个用于创建Observable的工厂函数
3. 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)

12. To

将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 介绍与详解实例

实例代码：

• 代码下载
  
• Github
  
• Gitee（码云）