0

我正在努力理解rxJsflatMap和concatMaprxJs 之间的区别。

我能理解的最明确的答案是这里的区别-concatmap-and-flatmap

所以我自己去尝试了一些东西。

import "./styles.css";
import { switchMap, flatMap, concatMap } from "rxjs/operators";
import { fromFetch } from "rxjs/fetch";
import { Observable } from "rxjs";

function createObs1() {
  return new Observable<number>((subscriber) => {
    setTimeout(() => {
      subscriber.next(1);
      subscriber.complete();
    }, 900);
  });
}

function createObs2() {
  return new Observable<number>((subscriber) => {
    setTimeout(() => {
      subscriber.next(2);
      //subscriber.next(22);
      //subscriber.next(222);
      subscriber.complete();
    }, 800);
  });
}

function createObs3() {
  return new Observable<number>((subscriber) => {
    setTimeout(() => {
      subscriber.next(3);
      //subscriber.next(33);
      //subscriber.next(333);
      subscriber.complete();
    }, 700);
  });
}

function createObs4() {
  return new Observable<number>((subscriber) => {
    setTimeout(() => {
      subscriber.next(4);
      subscriber.complete();
    }, 600);
  });
}

function createObs5() {
  return new Observable<number>((subscriber) => {
    setTimeout(() => {
      subscriber.next(5);
      subscriber.complete();
    }, 500);
  });
}

createObs1()
  .pipe(
    flatMap((resp) => {
      console.log(resp);
      return createObs2();
    }),
    flatMap((resp) => {
      console.log(resp);
      return createObs3();
    }),
    flatMap((resp) => {
      console.log(resp);
      return createObs4();
    }),
    flatMap((resp) => {
      console.log(resp);
      return createObs5();
    })
  )
  .subscribe((resp) => console.log(resp));

console.log("hellooo");

我在这里用过那个游乐场的例子

问题

1） 据我了解，flatMap 的使用应该混合输出，以便控制台日志类似于（1、3、2、4、5）。我已经尝试了 30 多次并且总是在同一行（1、2、3、4、5）

我做错了什么或有什么不明白的错误？

2） 如果打开createObs2()并且createObs3()您删除注释并包含具有多个发出事件的代码，那么事情就会变得混乱。即使您更改为 concatMap ，它也会使事情变得混乱，结果也很复杂。我期望只出现一次的多个数字多次出现。结果可能是 (1, 2, 33, 3, 2, 22, 3, 33, 4, 5, 4, 3, 4, 5) 为什么会这样？

我如何在操场上测试这个例子。我只从最后一个 console.log("hello") 中删除了 1 个字母。只有一个更改，例如 console.log("heloo") 然后被观察并再次编译项目并在控制台中打印输出。

编辑：我去 flatMap 和 concatMap 的原因是使用 http 库找到一个替代嵌套订阅的角度。

createObs1().subscribe( (resp1) => {
          console.log(resp1);
          createObs2().subscribe( (resp2) => {
               console.log(resp2);
               createObs3().subscribe( (resp3) => {
                   console.log(resp3);
                   createObs4().subscribe( (resp4) => {
                        console.log(resp4);
                        createObs5().subscribe( (resp5) => {
                            console.log(resp5);
                            })
                        })
                   })
               })
             })
4

2 回答 2

1

您的测试场景不足以看到这两个运算符之间的差异。在您的测试用例中，每个 observable 只发出 1 次。concatMap如果一个 observable 只发出一个值，那么和flatMap（又名 mergeMap）之间真的没有什么不同。只有在多次排放时才能看到差异。

所以，让我们使用不同的场景。让我们有一个source$observable，它只是每 1 秒发出一个递增的整数。然后，在我们的“高阶映射运算符”（concatMap& mergeMap）中，我们将返回一个每 1 秒发出可变次数的可观察对象，然后完成。

// emit number every second
const source$ = interval(1000).pipe(map(n => n+1)); 

// helper to return observable that emits the provided number of times
function inner$(max: number, description: string): Observable<string> {
  return interval(1000).pipe(
    map(n => `[${description}: inner source ${max}] ${n+1}/${max}`),
    take(max), 
  );
}

然后让我们基于source$和定义两个独立的可观察对象inner$；一用concatMap一用flatMap，观察输出。

const flatMap$ = source$.pipe(
    flatMap(n => inner$(n, 'flatMap$'))
);

const concatMap$ = source$.pipe(
    concatMap(n => inner$(n, 'concatMap$'))
);

在查看输出的差异之前，让我们先谈谈这些运算符的共同点。他俩：

  • 订阅传入函数返回的 observable
  • 从这个“内部可观察”排放出排放物
  • 取消订阅内部 observable(s)

不同的是，他们如何创建和管理内部订阅：

concatMap- 一次只允许一个内部订阅。当它接收到发射时，它一次只会订阅一个内部可观察对象。所以它最初会订阅“emission 1”创建的 observable，只有在它完成后，才会订阅“emission 2”创建的 observable。这与concat静态方法的行为方式一致。

flatMap( akamergeMap ) - 允许许多内部订阅。因此，它会在接收到新的发射时订阅内部的 observables。这意味着发射不会按任何特定顺序进行，因为它会在其任何内部可观察对象发射时发射。这与merge静态方法的行为方式一致（这就是我个人更喜欢名称“mergeMap”的原因）。

这是一个StackBlitz，它显示了上述 observables 的输出concatMap$和mergeMap$： concatMap 与 mergeMap

希望以上解释有助于解决您的问题！

#1 - “使用 flatMap 应该混合输出”

这没有像您预期的那样工作的原因是因为只有一个发射通过flatMap，这意味着您只有一个“内部可观察”发射值。如上例所示，一旦 flatMap 接收到多个发射，它就可以有多个独立发射的内部 observable。

#2 - “ ...并包含带有多个发出事件的代码，然后事情变得一团糟。 ”

“事情变得一团糟”是由于有多个发出值的内部订阅。

对于您提到的关于使用concatMap并仍然获得“混合”输出的部分，我不希望这样。当启用“自动保存”时，我在 StackBlitz 中看到了具有可观察到的排放的奇怪行为（似乎有时它没有完全刷新并且旧订阅似乎在自动刷新后仍然存在，这会产生非常混乱的控制台输出）。也许代码沙箱也有类似的问题。

#3 - “我去 flatMap 和 concatMap 的原因是使用 http 库找到一个替代嵌套订阅的角度”

这是有道理的。你不想搞乱嵌套订阅，因为没有一种很好的方法可以保证内部订阅会被清理。

在大多数使用 http 调用的情况下，我发现这switchMap是理想的选择，因为它会减少您不再关心的内部可观察对象的排放。想象一下，您有一个id从路由参数中读取的组件。它使用它id来进行 http 调用以获取数据。

itemId$ = this.activeRoute.params.pipe(
    map(params => params['id']),
    distinctUntilChanged()
);

item$ = this.itemId$.pipe(
    switchMap(id => http.get(`${serverUrl}/items/${id}`)),
    map(response => response.data)
);

我们只想item$发出“当前项目”（对应于 url 中的 id）。假设我们的 UI 有一个按钮，用户可以单击该按钮导航到下一个项目，id并且您的应用程序发现自己有一个喜欢点击的用户，该用户不断破坏该按钮，这比 http 调用返回数据的速度更快地更改了 url 参数。

如果我们选择mergeMap，我们最终会得到许多内部可观察对象，它们会发出所有这些 http 调用的结果。充其量，屏幕会随着所有这些不同的呼叫回来而闪烁。在最坏的情况下（如果调用无序返回），UI 将显示与 url 中的 id 不同步的数据:-(

如果我们选择concatMap，用户将被迫等待所有的 http 调用依次完成，即使我们只关心最近的一个。

但是，对于switchMap，每当收到新的发射 ( itemId) 时，它将取消订阅前一个内部可观察对象并订阅新的。这意味着它永远不会发出不再相关的旧 http 调用的结果。:-)

需要注意的一点是，由于 http observables 只发出一次，因此各种运算符 ( switchMap, mergeMap, concatMap) 之间的选择似乎没有什么区别，因为它们都为我们执行“内部 observable 处理”。但是，如果您开始接收多个发射，最好对您的代码进行未来验证并选择真正为您提供所需行为的代码。

于 2021-02-23T06:13:45.360 回答
0

每次第一个 observable 发射时，都会在 中创建第二个 observableflatMap并开始发射。但是，第一个 observable 的值不会进一步传递。

每次第二个 observable 发出时，下一个flatMap会创建第三个 observable，依此类推。同样，进入的原始值flatMap不会进一步传递。

createObs1()
  .pipe(
    flatMap(() => createObs2()), // Merge this stream every time prev observable emits
    flatMap(() => createObs3()), // Merge this stream every time prev observable emits
    flatMap(() => createObs4()), // Merge this stream every time prev observable emits
    flatMap(() => createObs5()), // Merge this stream every time prev observable emits
  )
  .subscribe((resp) => console.log(resp));

// OUTPUT:
// 5

因此，只有从createObs5()实际发出的值才会发送给观察者。先前的 observable 发出的值刚刚触发了新 observable 的创建。

如果您要使用merge，那么您将得到您可能期望的结果：

createObs1()
  .pipe(
    merge(createObs2()),
    merge(createObs3()),
    merge(createObs4()),
    merge(createObs5()),
  )
  .subscribe((resp) => console.log(resp));

// OUTPUT:
// 5
// 4
// 3
// 2
// 1
于 2021-02-22T10:24:09.953 回答