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

咱们知道在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) 例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$完成后再订阅执行promiseB$，这也是concat的主要特色。

concatAll

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

stackblitz 例 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$, "PromiseA is been Subscribed", promiseB$ 被顺序同时打印出来
2. 一秒后, " PromiseA ", " PromiseB is been Subscribed " 也被顺序同时打印出来
3. 又过了一秒，"PromiseB"被最后打印

解释下各个步骤

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

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

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

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

如今咱们但愿这些数据流有一个上下游关系，后面的数据流能接受前面的数据并和本身产生的数据进行再加工，再抛给下游，这时就须要使用到concatAll，来看下面这个使用concatAll后的改进版。

stackblitz 例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$这个数据流汇入到promiseA$的过程，咱们仔细来看:

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

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

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

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

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

concatMap

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

stackblitz 例 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 例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就好了。

做者说

这是我第一篇公开的文章， 文章里面有什么不对的欢迎你们挑刺~