rxJs中的flatMap与concatMap之争

Question

我正在努力理解rxJs中的平面图和连续图之间的区别。

我能理解的最清楚的答案是concatmap和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）根据我的理解，平面图的使用应该混合输出，这样控制台日志就像（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)为什么会发生这种情况？

我是如何在操场上测试这个例子的。我只从最后一个控制台中删除了1个字母。日志（“你好”）。只有一个更改，例如console。然后观察，再次编译项目，并在控制台中打印输出。

编辑：我之所以选择flatMap和concatMap，是为了使用http库在angular中找到嵌套订阅的替代品。

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);
                            })
                        })
                   })
               })
             })

Answer 1

每次第一个可观察对象发出时，都会在平面图中创建第二个可观察对象并开始发出。但是，第一个可观察对象的值不会进一步传递。

每次第二个可见光发射时，下一个平面图就会创建第三个可见光，依此类推。同样，进入平面图的原始值不会进一步传递。

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（）发出的值才能真正发送给观察者。先前观测值发出的值刚刚触发了新观测值的创建。

如果要使用“合并”（merge），则会得到您可能期望的结果：

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

// OUTPUT:
// 5
// 4
// 3
// 2
// 1

Answer 2

您的测试场景实际上不足以看到这两个操作符之间的差异。在您的测试用例中，每个可见光仅发射1次。如果一个可观察对象只发出一个值，那么concatMap和flatMap（又称mergeMap）之间实际上没有什么区别。只有当存在多重排放时，才能看到差异。

所以，让我们使用不同的场景。让我们有一个source$可观察的，它只是每1秒发出一个递增的整数。然后，在我们的“高阶映射运算符”（concatMap

// 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), 
  );
}

然后，让我们基于源$和内部$定义两个单独的观测值；一个使用concatMap，另一个使用flatMap，观察输出。

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

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

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

不同之处在于，它们是如何创建和管理内部订阅的：

一次只允许一个内部订阅。当它接收到辐射时，它一次只能订阅一个内部可见光。因此，它将首先订阅由“发射1”创建的可观测值，并且只有在完成后，它才会订阅由“发射2”创建的可观测值。这与concat静态方法的行为一致。

flatMap（又名mergeMap）允许许多内部订阅。因此，当接收到新的排放物时，它将订阅内部观测值。这意味着发射将不会以任何特定的顺序进行，因为它将在其任何内部可观测物发射时发射。这与静态方法的行为一致（这就是为什么我个人更喜欢“mergeMap”这个名称）。

希望上面的解释能帮助你澄清问题！

#1-"平面图的使用应该混合输出"

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

#2-“…并将代码包含在多个发出的事件中，那么事情就会变得一团糟。”

“事情变得混乱”是由于有多个内部订阅发出值。

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

#3-“我去了平面图和concatMap的原因是使用超文本传输协议库找到角中嵌套订阅的替代品”

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

在大多数使用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)
);

我们希望只发出“当前项”（对应于url中的id）。假设我们的UI有一个按钮，用户可以通过点击id导航到下一个项目，你的应用程序会发现一个喜欢点击的用户不断点击该按钮，这会改变url参数，甚至比http调用返回数据的速度还要快。

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

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

但是，使用switchMap，每当接收到新的排放（itemId）时，它将取消订阅以前的内部可观测数据并订阅新的。这意味着它将永远不会发出不再相关的旧http调用的结果。：-）

需要注意的一点是，由于http observables只发出一次，所以在各种操作符（switchMap，mergeMap，concatMap）之间的选择似乎没有什么区别，因为它们都为我们执行“内部可观察处理”。然而，如果您开始接收到不止一次的发射，最好是对您的代码进行未来验证，并选择一个真正提供您想要的行为的代码。