如何迭代返回RxJS Observable的异步函数并在返回false时终止？

嘿，作为RxJS新手，你遇到的这个问题其实很典型——咱们先拆解下代码里的两个核心问题：

1. 所有验证函数同时跑起来了：你用from把三个验证函数放进流里，再用map(data => data())调用它们。但from是同步发射数组里的所有函数的，这就导致三个验证会同时执行，完全不是你想要的“逐个迭代、失败即停”的逻辑，哪怕validateC返回false，前两个可能已经跑完了。
2. 流的层级没处理对：map(data => data())之后，你的流变成了Observable<Observable<any>>（Observable嵌套Observable），这时候takeWhile判断的根本不是验证结果里的result，而是Observable对象本身，自然起不到停止的作用。

修正后的代码方案 #

咱们用concatMap就能一次性解决这两个问题，它既能保证验证函数依次执行（等前一个完成再跑下一个），还能自动把嵌套的Observable扁平化，直接拿到验证结果：

import { Observable, from, of } from 'rxjs';
import { concatMap, takeWhile } from 'rxjs/operators';

function validateA(): Observable<any> { 
  return of({ id: "A", result: true }); 
}
function validateB(): Observable<any> { 
  return of({ id: "B", result: true }); 
}
function validateC(): Observable<any> { 
  return of({ id: "C", result: false });
}

from([validateA, validateB, validateC])
  .pipe(
    // 逐个执行验证函数，等前一个完成再处理下一个，同时扁平化结果
    concatMap(validateFn => validateFn()),
    // 当result为false时停止，第二个参数true让失败的那个结果也能被捕获到
    takeWhile(data => data.result, true)
  )
  .subscribe({
    next: data => {
      if (data.result) {
        console.log(`${data.id} passed!`);
      } else {
        console.log(`${data.id} failed! 停止后续验证～`);
      }
    },
    complete: () => console.log('验证流程结束')
  });

为啥这么改？ #

• concatMap：替代原来的map，它会严格按照数组顺序处理每个验证函数，必须等前一个验证的Observable完成，才会调用下一个函数，完美实现“迭代执行”的需求。
• takeWhile：现在它接收的是每个验证返回的真实数据，当data.result为false时，会立刻终止整个流，不会再执行后续的验证。第二个参数true是个小技巧，能让导致停止的那个失败结果也被发射出来，方便你做失败提示。

运行这段代码后，你会看到符合预期的输出：

A passed!
B passed!
C failed! 停止后续验证～
验证流程结束

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