Rxjava2如何简洁地将结果从前一个节点返回到链中的下一个节点

Posted 2023-03-08

【中文标题】Rxjava2如何简洁地将结果从前一个节点返回到链中的下一个节点

【英文标题】：Rxjava2 how to return result from previous node to the next node in the chain succinctly

【发布时间】：2019-01-12 05:16:08

【问题描述】：

我在android中使用rxjava2，有时会遇到这样的问题：

Observable.fromArray(
         // maybe a list about photo url in SDCard
    )
    .flatMap(
         // make list to single and upload to server,
         // use retrofit observable<uploadResponse>
    )
    .map(uploadResponse -> 
        // *Problem of point
    )
    .map(
        // in this map I want to get this single photo url in SDCard
        // and some info from uploadResponse,how to do in 
        // *Problem of point of previous map?
     );

这段代码忽略了一些线程切换和其他不重要的步骤。这个问题也让我对nodejs的承诺感到困惑，如何将一些值传递给链的每一步？我在firebase上使用nodejs的版本6，它不支持等待。

【参考方案1】：

至于javascript部分的问题，用async..await方便地解决了：

(async () => 
  const foo = await fooPromise;
  const bar = await getBarPromise(foo);
  const baz = await getBazPromise(bar);
)()
.catch(console.error);

我在firebase上使用nodejs的version6，它不支持await。

async..await 背后的模式自从 Node 4 中出现生成器以来就被广泛使用。同样的事情可以用最著名的实现 co 来实现：

const co = require('co');

co(function * () 
  const foo = yield fooPromise;
  const bar = yield getBarPromise(foo);
  const baz = yield getBazPromise(bar);
)
.catch(console.error);

由于async 函数是promise 的语法糖，async 函数所做的一切都可以单独用promise 重写。

鉴于以后可能需要一个值 (foo)：

fooPromise
.then(foo => getBarPromise(foo))
.then(bar => getBazPromise(bar))
.then(baz => 
  // foo and baz are needed here
);

它应该与其他结果一起通过链：

fooPromise
.then(foo => 
  return getBarPromise(foo).then(bar => ( foo, bar ));
)
.then(( foo, bar ) => 
  return getBazPromise(bar).then(baz => ( foo, baz ));
)
.then(( foo, baz ) => 
  // foo and baz are destructured from the result
);

或者应该嵌套承诺以使foo 在当前函数范围内可用：

fooPromise
.then(foo => 
  return getBarPromise(foo)
  .then(bar => getBazPromise(bar))
  .then(baz => 
    // foo and baz are available here
  )
);

相同的配方适用于其他 API 和语言。例如，RxJS：

fooObservable
.flatMap(foo => Observable.zip(
  Observable.of(foo),
  getBarObservable(foo)
))
.map(([foo, bar]) => 
  // foo and bar are destructured from the result
)

【讨论】：

我现在的代码和你的思路是一样的，但是我觉得这种方式类似于回调。如果需要持有更多的值并传递越来越深的层次（js的then或java的map），就会出现回调地狱一样的东西。

在rxjava中，我使用了compose操作符，它调用了zip方法的wrap(uploadResponse+pic local Url)，但是这种用法会导致不好的情况，比如我的上传图片功能。当我包装单张图片的上传时，我需要在其中添加一个 zip 返回只是为了上传许多图片功能，但可能其他功能只想获取上传响应，所以我的上传单张图片功能不能在其他地方重复使用

多层嵌套不太方便，但它们几乎不是回调地狱，因为后者意味着几乎无限的嵌套层级。特定于可观察对象的事情是链部分可以定义为变量并在以后组合在一起，因为它们在订阅之前不会被触发，而这种方法不适用于承诺。我不确定 RxJava，但在 RxJS 6 中，可以提取管道并将其用作自定义运算符 let customPipeline = pipe(map(...), map(...)); ...; observable.pipe(customPipeline)。这消除了嵌套回调。

无论如何，这些都是这里的选项。您可以嵌套链的一部分，或者通过传递链的某些部分中未使用的数据来展平链。重用代码的效率取决于您。在上面的代码中，它将是.map(([foo, bar]) => [foo, mapFnThatNeedsBar(bar)])。