Rxjs中concat, concatAll, concatMap及concatMapTo的理解及应用

ゝ一世哀愁。 2023-07-04 03:29 9阅读 0赞

我们知道在Rxjs中以concat开头的操作符都是用于合并数据流的，它的特点就是将前后两个数据流串起来，类似于Array.concat方法。 concat家族中有concat, concatAll, concatMap, concatMapTo等操作符，我们来依次比较这些操作符的区别及应用。

### concat ###

首先concat可以简单的将两个数据流前后收尾相接的串起来,例如

例1-1

// RxJS v6+
    import  {  of, concat }  from  'rxjs';
    concat(
      of(1,  2,  3),
      // subscribed after first completes
      of(4,  5,  6),
      // subscribed after second completes
      of(7,  8,  9)
    )
    // log: 1, 2, 3, 4, 5, 6, 7, 8, 9
    .subscribe(console.log);
    复制代码

代码会按序输出， 这是一个同步的例子，我们再举一个异步的例子

[(stackblitz)][stackblitz] 例1-2

import { concat, merge, defer, from } from 'rxjs'; 
    
    console.log('Start')
    const promiseA$ = defer(()=>from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove('PromiseA')
      }, 1000)
    })))
    const promiseB$ = defer(()=>from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove('PromiseB')
      }, 1000)
    })))
    
    // 会依次间隔一秒打印Start, PromiseA, PromiseB
    concat(promiseA$, promiseB$).subscribe(x => console.log(x));
    复制代码

在这个例子中，会每间隔一秒依次打印Start, PromiseA和PromiseB, 即concat会要等前一个promiseA ，这也是concat的主要特点。

### concatAll ###

接下来介绍concatAll, 我们知道带有All的是高阶Observable操作符, concatAll就是concat的处理高阶Oberservable的操作符，这个操作符有哪些特点呢？我们改写下上一个例子(例1-2)来看下concatAll是怎么处理高阶Oberservable的。

[stackblitz][stackblitz 1] 例 2-1

import { concat, defer, from, of } from 'rxjs'; 
    import { tap, concatAll } from 'rxjs/operators';
    
    console.log('Start')
    
    const promiseA$ = defer(()=>from(new Promise((reslove, reject)=>{
      console.log('PromiseA is been Subscribed ')
      setTimeout(()=>{
        reslove('PromiseA')
      }, 1000)
    })))
    
    const promiseB$ = defer(()=>from(new Promise((reslove, reject)=>{
      console.log('PromiseB is been Subscribed ')
      setTimeout(()=>{
        reslove('PromiseB')
      }, 1000)
    })))
    
    // 会依次间隔一秒打印Start, PromiseA, PromiseB
    of(promiseA$, promiseB$).pipe(tap(console.log),concatAll()).subscribe(x => console.log(x));
    复制代码

这个例子结果和例1-2是一样的，但过程有些不同，为了便于观察，这里加了很多console.log，我们看下这段代码的执行结果

1.  首先 "Start", promiseA 被顺序同时打印出来
2.  一秒后, " PromiseA ", " PromiseB is been Subscribed " 也被顺序同时打印出来
3.  又过了一秒，"PromiseB"被最后打印

解释下各个步骤

1.   *  of先会把第一个数据流promiseA 数据流作为一个数据被concatAll接受并订阅，所以"PromiseA is been Subscribed"被打印出来；
     *  这时由于promiseA$是异步数据流，它还没有完结，因此它内部的数据暂时不会向下游抛出；
     *  接着of向下游抛出了下一个数据流promiseB 这个数据流并把它暂存了起来。
2.  一秒之后, concatAll中的第一个数据流promiseA 没有被订阅，因此订阅了它并导致" PromiseB is been Subscribed "被打印
3.  再一秒后，promiseB$完成

一句话总结, concatAll顺序接受上游抛出的各个数据流作为它的数据， 若前面的数据流不能同步的完结，它会暂存后续数据流，当前数据流完成后它才会订阅后一个暂存的数据流

### 使用map+concatAll在数据上衔接数据流 ###

上面两个例子(例1-2, 例2-1)都有一个问题，就是后一个数据流拿不到前一个数据流抛出的数据，这时因为 concat方法接收的参数或者concatAll接收的数据都是各种数据流，这些数据流在数据传递上是并行关系，不是上下游关系，只是在执行顺序上是前后关系 , 所以这些数据流完成后会直接把数据抛给下游 。

现在我们希望这些数据流有一个上下游关系，后面的数据流能接受前面的数据并和自己产生的数据进行再加工，再抛给下游，这时就需要使用到concatAll，来看下面这个使用concatAll后的改进版。

[stackblitz][stackblitz 2] 例3-1

import { concat, defer, from } from 'rxjs'; 
    import { concatAll, map, tap } from 'rxjs/operators'; 
    
    console.log('Start')
    
    const promiseA$ = defer(()=>from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove('PromiseA')
      }, 1000)
    })))
    
    // 这是一个会返回数据流promiseB$的函数
    const promiseB = data => from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove(`${data} then PromiseB`)
      }, 1000)
    }))
    
    // map会将把上游完成后的数据通过promiseB转换成promiseB$数据流
    // 并传递给concatAll, concatAll将promiseB$连接下游数据流
    // 这里将在两秒后打印出 PromiseA then PromiseB
    promiseA$.pipe(
      map(promiseB),
      concatAll()
    ).subscribe(x => console.log(x))
    复制代码

可以看到这个例子的写法与之前例子的写法有较大的不同，各数据流不是一个并行的写法，而是有一个链式的前后关系。

promiseB是一个会返回from数据流的函数, 为了描述方便, 我们现在把promiseB这个方法返回的数据流称为promiseB$。

在这个例子中其实就是一个把promiseB 的过程，我们仔细来看:

1.  promiseA$本身就是一个异步数据流, 它将在1s后完成并执行下游
2.  map获取上游的数据并将其转换成一个包含上游数据的数据流promiseB 抛给了下游的的concatAll
3.  这时对concatAll来说它接收到的数据就是数据流promiseB ，并将其完成后的数据抛给下游。

可以这么理解, map+concatAll的组合就是帮助promiseB$衔接上下游的

OK，我们用一张手画伪弹珠草图来描述下上面这个例子的区别

![format_png][]

可以看到concat中 数据是不会在各个子数据流中传递的，统一抛给了下游；

在concatAll的图例中，map将B作为数据抛给了concatAll，B在concatAll中被订阅, 完成后产生的数据直接抛给了下游。

### concatMap ###

为了将两个数据流在数据上进行衔接需要用到map+concatAll这个固定的组合，有没有方法能简洁的就把这个事干了的呢？那就是concatMap, map+concatAll的语法糖。

[stackblitz][stackblitz 3] 例 4-1

import { concat, defer, from } from 'rxjs'; 
    import { concatMap, map, tap } from 'rxjs/operators'; 
    
    console.log('Start')
    
    const promiseA$ = defer(()=>from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove('PromiseA')
      }, 1000)
    })))
    
    // 这是一个会返回数据流promiseB$的函数
    const promiseB = data => new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove(`${data} then PromiseB`)
      }, 1000)
    })
    
    // concatMap 可以接收一个返回Promise的函数或者是数据流
    // 这里将在两秒后打印出 PromiseA then PromiseB
    promiseA$.pipe(
      concatMap(promiseB)
    ).subscribe(x => console.log(x))
    复制代码

这个例子中用cancatMap代替了 map+concatAll, 可以看到效果是一样的。

值得注意的是我们发现promiseB函数和上面的例子（例3-1）也产生了不同，当然promiseB如果和例3-1是一样的话在这里也正常能运行。在这里promiseB去掉了defer和from的包裹，这是因为concatMap可以接受一个Promise返回作为一个新的数据流。这样代码会比之前的例子更加简洁一点。

第二个参数

concatMap还可以接受第二个参数, 第二个参数是一个回调函数, 这个回调函数会在当前数据流完成后被立即调用。这个回调函数接受的第一个参数是当前数据流接受的数据参数，第二个参数是当前数据流处理后返回的数据。

考虑这个场景，我们有一个页面，页面初始化的时候有个初始化接口获取数据，页面上有个下一步按钮，这个按钮触发的事件需要使用初始化接口的数据。因为事件流返回的数据就是事件本身，此时我们不需要这个数据，我们只需事件发生后获取之前抛出的数据，这时我们可以使用第二个回调函数：

[stackblitz][stackblitz 4] 例4-2

import { concat, defer, from, fromEvent } from 'rxjs'; 
    import { tap, concatMap } from 'rxjs/operators'; 
    
    // 使用promise模拟数据请求过程
    const req$ = defer(()=>from(new Promise((reslove, reject)=>{
      setTimeout(()=>{
        reslove('This is init data')
      }, 1000)
    })))
    
    // 事件流
    const button$ = _ => fromEvent(document.getElementById('button'), 'click')
    
    // 点击按钮后输出请求内容
    // 这里会打印出 This is init data
    req$.pipe(
      concatMap(button$, (data, event) => data)
    ).subscribe(x => console.log(x))
    复制代码

可以看到点击按钮后会输出请求数据，而参数event就是事件流返回的点击事件本身了。

### concatMapTo ###

如果说cancatMap相当于map+concatAll, 那concatMapTo就相当于mapTo + concatAll了，就是把上游的数据统一映射成下游的数据。

对于concatMapTo需要注意这些

1.  concatMapTo接受的第一个参数和concatMap不同， 它直接接受一个数据流，并把这个数据流产生的数据直接抛给下游
2.  它的第二个参数与concatMap相同

所以对于上一个例子（例4-2）我们可以有个小改进，就是把button$外围包裹的函数去掉，直接把fromEvent给concatMapTo就行了。

[stackblitz]: https://stackblitz.com/edit/rxjs-kw4roh?devtoolsheight=60
[stackblitz 1]: https://stackblitz.com/edit/rxjs-uprdaj?devtoolsheight=60
[stackblitz 2]: https://stackblitz.com/edit/rxjs-rt6pez?devtoolsheight=60
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcwLnR1aWNvb2wuY29tL3JRVnIybXYuanBnIXdlYg?x-oss-process=image/format,png
[stackblitz 3]: https://stackblitz.com/edit/rxjs-bxiuju?devtoolsheight=60
[stackblitz 4]: https://stackblitz.com/edit/rxjs-d3fyj4