使用多播重复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)));

或带有repeatWhen（）的动态延迟值。

这在普通（不可连接）可观察的情况下可以正常工作，但在多播情况下则不行。

代码示例：

作品

    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

仍然不清楚这是否是你想要的，但也许这会帮助你

  Observable<Integer> integerObservable1;
@Override
public void run(String... args) throws Exception {

    Integer[] integers = {1, 2, 3};
    Observable<Integer> integerObservable = Observable.fromArray(integers);

    integerObservable1 = Observable.zip(integerObservable.observeOn(Schedulers.newThread()).delay(100, TimeUnit.MILLISECONDS), integerObservable.observeOn(Schedulers.newThread()).delay(200, TimeUnit.MILLISECONDS), (integer, integer2) -> {
        System.out.println(integer + " " + integer2);
        return integer;
    })
            .doOnComplete(() -> {
                integerObservable1.subscribe();
            });
    integerObservable1.subscribe();

}

Answer 2

第二个示例中的问题不在多播中。它在collect操作符中，并在repeatWhen中。

当您连接到onComplete方法时，请考虑像“钩子”一样重复。这个“钩子”将截获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();

在本例中，我们使用observable的onComplete方法，我们按住计时器重置状态。当消费者使用dhe数据的速度慢于重复的延迟时，此示例没有考虑该选项（这将导致结果从一条数据链溢出到另一条数据链）。我建议使用其他方法来处理重复部分，例如：

最终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秒调用一次，它从一开始就重新启动整个管道。

希望这有帮助。