rxjs：从 2 个端点加载多对多并合并到单个 Observable答案

【问题标题】：rxjs: load many for many from 2 endpoints and merge into single Observablerxjs：从 2 个端点加载多对多并合并到单个 Observable
【发布时间】：2020-12-29 23:13:48
【问题描述】：

我有一个端点用于获取用户列表，另一个端点用于获取每个用户的公寓列表：

getUsers() => Observable<User[]>;
getUserApartments(userId) => Observable<Apartment[]>`

如何将两者中的数据合并为单个 observable：

const usersWithApartments$: Observable<{ user: User, apartments: Apartment[] }[]> = getUsers().pipe(
    // users.map(user => { user, apartments: getUserApartments(user.id) }) <-- turn this pseudocode into Rxjs
);

【问题讨论】：

标签： rxjs rxjs6 rxjs-observables rxjs-pipeable-operators

【解决方案1】：

由于getUsers() 的返回类型是Users[]，我假设所需的类型是{ user: User, apartments: Apartment[]> 的集合，而不是单个。

请考虑以下实现：

class UserApartments {
  user: User;
  apartments: Apartment[];
}

const usersWithApartments$: Observable<UserApartments[]> = getUsers()
  .pipe(
    switchMap(users => forkJoin(
      users.map(user => getUserApartments(user.id).pipe(
        map(apartments => ({ user, apartments }))
      ))
    ))
  );

【讨论】：

正确！在问题中修复它。并感谢您的解决方案！

【解决方案2】：

就是这样：

const toUserWithApartmentsStream = (user: User) => getUserApartments(user.id).pipe(
  map((apartments: Apartment[]) => ({ user, apartments }))
);

const usersWithApartments$: Observable<{ user: User, apartments: Apartment[] }[]> =
  getUsers().pipe(
    switchMap((users: User[]) => users),
    concatMap((user: User) => toUserWithApartmentsStream(user)),
    toArray()
  );

现在getUserApartments会一个接一个连续执行，我可以写mergeMap而不是concatMap，getUserApartments会同时执行。感谢 Rafi Henig 的评论

函数toUserWithApartmentsStream 只是一个帮助器，因此代码看起来更干净。我可以在concatMap 内内联写它。

【讨论】：

您的解决方案缺乏并发性
@RafiHenig 如果作者不想同时向 api 发送大量请求怎么办？我不会说这个答案缺乏并发性，我会说这个特定的答案将连续执行请求，以避免同时发送太多请求并使事情变得更糟。至于替代方案，我可以写mergeMap 而不是switchMap，我的答案将具有并发性。
@Goga Koreli 通常需要并发，此外mergeMap 不保证合并后的 observable 将按照与源中的项目到达相同的顺序发出 api 响应
我同意，通常是这样。在这个特定示例的实际场景中，如果有很多用户，则会导致同时发送大量网络调用，如果有多个经过身份验证的用户同时执行此操作，则会更多。因此很难打击后端。顺便说一句，我同意 mergeMap 不能保证到达的顺序，但是当结果在两种情况下都是数组时，我看不到有理由提及它。