如何序列化执行Observable数组，同时获取所有执行返回结果？

你可以通过 concat + toArray 的组合实现需求，既保证校验逻辑按顺序串行执行，又能收集所有行的校验结果，和你原来的逻辑完全兼容：

import { concat } from 'rxjs';
import { toArray, map } from 'rxjs/operators';

public validate():Observable<boolean>{
    // 生成冷Observable数组，订阅时才会触发校验逻辑
    const rowValidations:Observable<boolean>[] = dataRows.map(row=>this.validateSingleRow(row));
    return concat(...rowValidations).pipe(
      // 收集所有串行执行的结果为数组
      toArray(),
      // 原逻辑保持不变，判断所有校验是否都通过
      map(results=>results.every(r=>r))
    )
}

原理说明 #

• concat 会按传入顺序依次订阅内部 Observable，只有前一个 Observable 触发 complete 回调后，才会订阅下一个，天然保证了共享资源的串行访问，效果和Java的synchronized修饰方法一致。
• toArray 操作符会缓存 concat 流吐出的所有校验结果，等所有 Observable 都执行完成后，一次性输出结果数组，实现和 forkJoin 一致的结果收集能力。

注意事项 #

• 确保 validateSingleRow 返回的是会正常触发 complete 回调的冷 Observable，如果是不会结束的热 Observable，concat 会一直等待不会执行后续校验。
• 如果需要某行校验失败就立刻终止后续所有校验，节省资源，可以在 toArray 前添加 takeWhile(result => result, true) 操作符，只要某行返回false就会直接结束流。

等价写法 #

你也可以配合 from + concatMap 实现相同效果，写法更简洁：

import { from } from 'rxjs';
import { concatMap, toArray, map } from 'rxjs/operators';

public validate():Observable<boolean>{
    return from(dataRows).pipe(
      concatMap(row => this.validateSingleRow(row)),
      toArray(),
      map(results=>results.every(r=>r))
    )
}

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