如何在链接可观测对象时传递变量？

Question

我对RxJava非常陌生，每当我遇到需要将返回数据从一个可观察对象传递到“subscribe”调用为止的情况--我很难理解如何在没有任何补丁的情况下以“反应式”的方式完成这项工作……

例如：

Observable<GameObject> obs1 = func1();
Observable<GameObject> obs2 = func2();
Observable<GameObject> obs3 = func3();
Observable<GameObject> obs3 = func4();

我想发出obs1和obs2，得到它们的结果，然后发出obs3，然后是obs4，然后用subscribe结束链，同时可以访问obs1，obs2，obs3和obs4的结果。

调用的顺序很重要，我需要在执行obs3之前完成obs1和obs2。

对于obs3和obs4也是如此-我需要在执行obs4之前完成obs3。

我该怎么做呢？

我知道这是一个非常容易理解的问题--但当开发人员开始了解rxJava时，这是最有问题的问题之一。

谢谢。

Answer 1

您可以使用Observable.zip和简单的Observable.map/Observable.flatMap来完成

Observable.zip(obs1, obs2, (res1, res2) -> {
    // do stuff with res1, res2
    return obs3.flatMap(res3 -> {
        // do stuff with res1, res2, res3
        return obs4.flatMap(res4 -> {
            // do stuff with res1, res2, res3, res4
            return result;
        });
    });
});

这将强制您的日程安排要求：

• 可观察性1和2
• 可观察性3
• 可观察性4

Answer 2

因为不久前我也有过同样的疑问，所以这个问题看起来与观察物是如何工作的有关。

假设您使用以下内容创建了obs1和obs2：Observable<GameObject> obs1 = Observable.create(...) Observable<GameObject> obs2 = Observable.create(...)

您有两个独立且断开连接的流。当它们中的每一个都应该做一些事情时，比如网络请求或一些密集的后台处理，这就是你想要的，这可能需要一些时间。

现在，假设您想查看这两个结果，并在它们准备好时发出一个值(您没有明确地说明这一点，但这将帮助您理解它是如何工作的)。在这种情况下，您可以使用zipWith运算符，该运算符获取一对项目，第一个可观察对象中的第一个项目和第二个可观察对象中的第二个项目，将它们组合为单个项目，并将其发送到链中可能对其感兴趣的下一个项目。zipWith是从一个可观察对象中调用的，它希望将另一个可观察对象作为参数进行压缩。它还需要一个自定义函数，该函数知道如何压缩两个源项并从中创建一个新项。

Observable<CustomObject> obs3 = obs1.zipWith(obs2, new Func2<GameObject, GameObject, CustomObject>() {
    @Override 
    public CustomObject call(GameObject firstItem, GameObject secondItem) {
        return new CustomObject(firstItem, secondItem);
    }
});

在本例中，CustomObject只是一个pojo。但这可能是另一个长期运行的任务，或者是您需要对前两个可观察项目的结果做什么。

如果你想等待(或者，观察！)来自obs3的结果您可以在末尾插入另一个可见对象，该对象应该执行另一项处理。

Observable<FinalResult> obs4 = obs3.map(new Func1<CustomObject, FinalResult>() {
    @Override
    public FinalResult call(CustomObject customObject) {
         return new FinalResult(customObject);
    }
});

map操作符将一个对象转换(或映射)到另一个对象。因此，您可以执行另一段处理或任何数据操作，并从中返回结果。或者您的FinalResult可能是一个常规的类，比如CustomObject，只是保存对其他GameObject的引用。随便你怎么说。

根据您创建可观察对象的方式，它们可能尚未开始发射任何项。到目前为止，您只是创建和插入数据管道。为了触发第一个任务并使项目在流中流动，您需要订阅它。

obs4.subscribe();

总而言之，你并不是真的有一个单一的变量在整个链中传递。您实际上在第一个可观察对象中创建了一个项，当它准备就绪时通知第二个可观察对象，依此类推。此外，每个步骤(可观察的)以某种方式转换数据。所以，你有一系列的转换。

RxJava遵循函数方法，对数据应用高阶函数(map、zip、filter、reduce)。弄清楚这一点至关重要。此外，数据始终是不可变的:您不会真正更改可观察对象，也不会更改自己的对象。它会创建它们的新实例，而旧对象最终将被垃圾回收。因此，obs1.zip(...)不会更改obs1，它会创建一个新的可观察实例，您可以将其赋给一个变量。

您还可以删除变量赋值(obs1、obs2、obs3等)，只需将所有方法链接在一起即可。所有内容都是强类型的，因此编译器不会让您插入彼此不匹配的可观察对象(一个对象的输出应该与下一个对象的输入匹配)。

我希望它能给你一些想法！