多个连续的 Observable 依赖于单个 Observable

Question

我很好奇编写以下场景的更好方法。

假设我有一个

userObservable

从数据库获取用户数据（用户），那么我想

• 使用

  user

  、

  validateObservable

验证某些输入
• 使用

  user

  获取一些外部数据后，

  getExternalObservable

• 将externalData与

  user

  合并并再次更新数据库后，

  updateDbObservable

• 最后我想得到这个更新的用户数据。

通常我可以按如下方式执行此操作，但此代码看起来重复并生成有点难看的映射 -> forkJoin -> mergeAll 样板。

userObservable.pipe(
  map((user) => {
    return forkJoin({
      user: of(user),
      verifyResult: verifyObservable(user)
    });
  },

  mergeAll(),

  map(({ user }) => {
    return forkJoin({
      user: of(user),
      externalData : getExternalObservable(user)
    });
  }),
  
  mergeAll(),

  map(({user, externalData}) => {
    return updateDbObservable(user, externalData);
  }),

  mergeAll(),
);
    

谢谢你！

Answer 1

您可以极大地简化您的代码，但这取决于您使用这些可观察量的目标。您是否需要生成的可观察值发出某种形状？您需要重复使用这些数据吗？

---

筑巢

一种简化方法是像这样嵌套可观察量：

userObservable.pipe(
  switchMap(user => verifyObservable(user).pipe(
    switchMap(verifyResult => getExternalObservable(user).pipe(
      switchMap(externalData => updateDbObservable(verifyResult, externalData))
    ))
  ))
);

这里我们为

.pipe()

内的每个可观察值添加一个

switchMap

：

当

userObservable

发出时，结果流向

verifyObservable()

，然后流向

getExternalObservable()

，最后流向

updateDbObservable()

。

请注意，最内部的调用可以访问所有先前发出的值，因此无需使用

forkJoin

/

of

来构造对象。

---

单独的可观察值

如果您需要重用这些数据中的任何一个，那么声明单独的可观察量可能会很有用，如下所示：

const verifyUserResult = userObservable.pipe(
  switchMap(user => verifyObservable(user))
);

const externalData = userObservable.pipe(
  switchMap(user => getExternalObservable(user))
);

const updateDbResult = combineLatest([verifyUserResult, externalData]).pipe(
  switchMap(([verifyResult, externalData]) => updateDbObservable(verifyResult, externalData))
);

或更简单地说：

const verifyUserResult = userObservable.pipe(switchMap(verifyObservable));

const externalData = userObservable.pipe(switchMap(getExternalObservable));

const updateDbResult = combineLatest([verifyUserResult, externalData]).pipe(
  switchMap(data => updateDbObservable(...data))
);

通过这段代码，我们定义了可以在代码中其他地方轻松引用的小型可观察量。可重用性是一个好处，但另一个好处也是可读性。

注意

updateDbResult

的定义如何更容易在心理上解析，您可以看到它需要两个可观察的输入并调用

updateDbObservable()

。您无需关心每个来源的内部细节。

共享订阅

通常，如果您多次订阅同一个可观察对象，您将希望共享相同的订阅，这样就不会为单独的订阅者执行多个 http 调用。您可以使用

share

或

shareReplay

:

来实现此目的

const updateDbResult = combineLatest([verifyUserResult, externalData]).pipe(
  switchMap(([verifyResult, externalData]) => updateDbObservable(verifyResult, externalData)),
  shareReplay(1), // <--- all subscribers will share result instead of
);                //      executing a separate db update per subscribe