RxJava 在不重复执行的情况下组合可观察对象

Question

短篇小说：我有一种情况，我有 2 个具有单一用途的 Observable：

他们收到一些数据
他们return修改了数据
如果无法处理数据则抛出错误

他们各自负责处理不同类型的数据。另外我想在处理完两个数据后做一些事情。

我目前最好的实现如下，这些是我的 Observables：

    Single<BlueData> blueObservable = Single.create(singleSubscriber -> {
        if (BlueDataProcessor.isDataValid(myBlueData)) {
            singleSubscriber.onSuccess(BlueDataProcessor.process(myBlueData));
        }
        else {
            singleSubscriber.onError(new BlueDataIsInvalidThrow());
        }
    });

    Single<RedData> redObservable = Single.create(singleSubscriber -> {
        if (RedDataProcessor.isDataValid(myRedData)) {
            singleSubscriber.onSuccess(RedDataProcessor.process(myRedData));
        }
        else {
            singleSubscriber.onError(new RedDataIsInvalidThrowable());
        }
    });

    Single<PurpleData> composedSingle = Single.zip(blueObservable, redObservable,
            (blueData, redData) -> PurpleGenerator.combine(blueData, redData));

我还有以下订阅：

    blueObservable.subscribe(
            result -> {
                saveBlueProcessStats(result);
            },
            throwable -> {
                logError(throwable);
            });

    redObservable.subscribe(
            result -> {
                saveRedProcessStats(result);
            },
            throwable -> {
                logError(throwable);
            });


    composedSingle.subscribe(
            combinedResult -> {
                savePurpleProcessStats(combinedResult)
            },
            throwable -> {
                logError(throwable);
            });

我的问题：蓝色和红色数据被处理了两次，因为两个订阅都是运行我订阅了使用 Observable.zip().

创建的组合观察

如果不运行两次操作，我怎么能有这种行为？

Answer 1

这对于 1.x 中的 Single 是不可能的，因为没有 ConnectableSingle 的概念，因此没有 Single.publish。您可以通过 2.x 和 RxJava2Extensions 库实现效果：

SingleSubject<RedType> red = SingleSubject.create();
SingleSubject<BlueType> blue = SingleSubject.create();

// subscribe interested parties
red.subscribe(...);
blue.subscribe(...);

Single.zip(red, blue, (r, b) -> ...).subscribe(...);

// connect()
blueObservable.subscribe(blue);
redObservable.subscribe(red);

RxJava 在不重复执行的情况下组合可观察对象

RxJava combining observables without repeating execution

java

rx-java

rx-android

rx-java2