如何在RxJS concatMap执行完成后发起后续API调用

核心原理 #

RxJS 中，当源 Observable（本例中为from(array)）的所有元素都发射完毕、所有内部被订阅的 Observable 都执行完成后，会自动触发流的完成通知。而concatMap本身就具备串行执行的特性：必须等上一个内部API请求完成后，才会发起下一个请求，因此流完成通知的触发时机，就恰好是所有串行API调用全部执行完毕的节点。

实现方式 #

方式1：在订阅的完成回调中触发额外API #

这是最直观、最容易调试的实现方式，不需要额外拼接流逻辑。
原来的订阅逻辑目前只处理了next返回的响应，只需要新增complete回调的实现即可——这个回调只有在整个流正常执行完成、没有未处理的并发请求时才会触发：

// 提前定义好需要在全流程结束后执行的额外API逻辑
const runExtraApi = () => {
  // 替换为你的实际额外API调用代码
  return AnotherCustomApiRequest(/* 可传入前面收集的responses数组作为参数 */)
    .subscribe(extraRes => {
      // 处理额外API的响应结果
    })
}

from(array)
  .pipe(
    // 修正原代码中箭头函数的冗余return写法
    concatMap(el => APICall(el))
  )
  .subscribe({
    next: response => { 
      if(!(response instanceof HttpErrorResponse)) {
        responses.push(response);
      }
    },
    // 所有串行请求全部执行完成后，才会进入这个回调
    complete: () => {
      runExtraApi()
    },
    error: err => {
      // 可选：增加流程异常的兜底处理
      console.error('串行处理流程中断', err)
    }
  })

这种方式的优势是可以非常方便的把前面已经收集到的responses数组作为参数传入后续的额外API调用，适配需要依赖前面处理结果的场景。

方式2：通过流操作符把额外API接入链式流程 #

如果希望整个逻辑保持链式的响应式写法，不把副作用写在订阅回调里，可以用concat操作符把原有处理流和额外API的流拼接起来。concat的执行规则是严格等前一个流完全完成后，才会订阅并执行下一个流，完全匹配时序要求：

import { from, concat } from 'rxjs'
import { concatMap } from 'rxjs/operators'

// 原有流程：串行跑完所有元素对应的API请求
const originalSerialFlow$ = from(array).pipe(
  concatMap(el => APICall(el))
)

// 额外API对应的请求流
const extraApiFlow$ = AnotherCustomApiRequest(/* 参数 */)

// 按顺序拼接两个流：原流程跑完，才会发起额外API请求
concat(originalSerialFlow$, extraApiFlow$)
  .subscribe({
    next: response => {
      // 这里会依次收到所有原流程的响应，最后收到额外API的响应
      if(!(response instanceof HttpErrorResponse)) {
        responses.push(response);
      }
    },
    error: err => {
      console.error('流程执行异常', err)
    }
  })

注意事项 #

• 注意区分concat和merge的执行逻辑：merge会同时订阅所有传入的流，会提前发起额外API调用，无法满足串行等待的时序要求。
• 如果流程中存在任意一个API请求抛出未捕获的错误，整个流会直接进入error分支，不会触发complete回调，此时如果需要无论正常完成还是出错都执行额外操作，可以在管道中加入finalize操作符替代complete回调。
• 原代码中concatMap(el => return APICall(el))的写法在箭头函数没有大括号时会触发语法错误，注意调整为符合箭头函数语法的返回写法。

内容的提问来源于stack exchange，提问作者YehorPavlenko