rx-java - 使用多播重复 RxJava 管道

Question

我是 RxJava 的新手，无法弄清楚如何使用 ConnectableObservable 实现可重复轮询，其中有 2 个订阅者在不同线程上处理事件。

我有一个大致如下所示的管道:

我想在延迟后以与 https://github.com/ReactiveX/RxJava/issues/448 中的解决方案类似的方式重复整个管道

Observable.fromCallable(() -> pollValue())
.repeatWhen(o -> o.concatMap(v -> Observable.timer(20, TimeUnit.SECONDS)));

或Dynamic delay value with repeatWhen() .

这适用于普通(不可连接的)Observable，但不适用于多播。

代码示例:

作品

    final int[] i = {0};
    Observable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++));

    integerObservable
            .observeOn(Schedulers.newThread())
            .collect(StringBuilder::new, (sb, x) -> sb.append(x).append(","))
            .map(StringBuilder::toString)
            .toObservable().repeatWhen(o -> o.concatMap(v -> Observable.timer(1, TimeUnit.SECONDS)))
            .subscribe(System.out::println);

不起作用:

    final int[] i = {0};
    ConnectableObservable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++))
            .publish();

    integerObservable.observeOn(Schedulers.newThread()).subscribe(System.out::println);

    integerObservable
            .observeOn(Schedulers.newThread())
            .collect(StringBuilder::new, (sb, x) -> sb.append(x).append(","))
            .map(StringBuilder::toString)
            .toObservable().repeatWhen(o -> o.concatMap(v -> Observable.timer(1, TimeUnit.SECONDS)))
            .subscribe(System.out::println);

    integerObservable.connect();

Answer 1

第二个例子中的问题不在于多播。它在收集运算符和 repeatWhen 中。

将 repeatWhen 视为连接到 onComplete 方法的“钩子(Hook)”。这个“钩子(Hook)”将拦截 onComplete 方法并“覆盖”它，因此它不会被调用，除非这个可观察对象在开始重复过程之前第一次完成。

如果没有调用 collect 运算符的 onComplete 方法，则不知道它应该收集多少项目。因此，您必须处理如何收集项目以及将它们存储在流外的位置的逻辑，但这将是一种解决方法。

这是一个例子:

    List<String> test = new ArrayList<>();
    final String[] currentString = {""};

    final int[] i = {0};
    ConnectableObservable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++))
                    .doOnComplete(() -> {
                        test.add(currentString[0]);
                        currentString[0] = "";
                    })
                    .repeatWhen(o -> {
                        return o.concatMap(v ->
                                Observable.timer(3, TimeUnit.SECONDS)
                                .doOnComplete(() -> {
                                    test.add(currentString[0]);
                                    currentString[0] = "";
                                }));
                    })
                    .publish();

    integerObservable
            .observeOn(Schedulers.newThread())
            .subscribe(System.out::println);

    integerObservable
            .observeOn(Schedulers.newThread())
            .map((sa) -> {
                currentString[0] = currentString[0] + sa;
                System.out.println(currentString[0]);
                return sa;
            })
            .subscribe();

在这个例子中，我们使用我们持有计时器的可观察对象的 onComplete 方法来重置我们的状态。当消费者消耗数据的速度比重复的延迟慢时，这个例子没有考虑这个选项(这将导致结果从一个数据链溢出到另一个数据链)。我建议使用其他方法来处理重复部分，例如:

最终 int[] i = {0};

    ConnectableObservable<Integer> integerObservable =
            Observable.defer(() -> Observable.fromArray(i[0]++, i[0]++, i[0]++)).publish();

    Observable b = integerObservable.observeOn(Schedulers.newThread());

    Observable a = integerObservable
            .observeOn(Schedulers.newThread())
            .collect(StringBuilder::new, (sb, x) -> sb.append(x).append(","))
            .map(StringBuilder::toString)
            .toObservable();

    Observable
            .interval(1, TimeUnit.SECONDS)
            .doOnSubscribe((d) -> {
                a.subscribe(System.out::println);
                b.subscribe(System.out::println);

                integerObservable.connect();
            })
            .doOnNext((d) -> {
                a.subscribe(System.out::println);
                b.subscribe(System.out::println);

                integerObservable.connect();
            })
            .doOnComplete(() -> {})
            .subscribe();

第一次连接是第一次执行，onNext 每 1 秒调用一次，它会从头开始重新启动整个管道。

希望这对您有所帮助。