设置mergeAll为10时，如何等待所有Observable执行完成？

核心思路 #

通过控制并发量的同时，用RxJS操作符收集所有请求结果，待全部任务完成后触发用户通知。推荐用mergeMap（并行控制）或concatMap（串行控制）结合toArray()实现，既解决高并发报错问题，又能统一等待所有请求结束。

并行处理方案（推荐） #

适合服务器能承受一定并发的场景，可自定义并发数：

import { from } from 'rxjs';
import { mergeMap, toArray, tap } from 'rxjs/operators';

// 模拟学生列表
const studentList = Array.from({ length: 200 }, (_, i) => ({ id: i, name: `学生${i+1}` }));

// 单个学生的API调用函数（替换成你的实际请求）
const createTestTask = (student) => {
  // 捕获单个请求的错误，避免整个流中断
  return fetch(`/api/create-test-task?studentId=${student.id}`)
    .then(res => res.json())
    .catch(err => ({ success: false, studentId: student.id, error: err.message }));
};

// 执行流程
from(studentList).pipe(
  // 控制并发数为20，可根据服务器性能调整
  mergeMap(student => createTestTask(student), 20),
  // 收集所有请求结果（成功/失败都包含）
  toArray(),
  // 可选：统计结果
  tap(allResults => {
    const successNum = allResults.filter(item => item.success !== false).length;
    const failNum = allResults.length - successNum;
    console.log(`任务分配完成：成功${successNum}个，失败${failNum}个`);
  })
).subscribe({
  next: (allResults) => {
    // 所有请求完成后通知用户
    alert('所有测试任务已分配完毕！');
    // 可在这里进一步处理结果，比如展示失败的任务列表
  },
  error: (err) => {
    console.error('全局错误：', err);
    alert('任务分配过程中出现未知错误，请重试');
  }
});

串行处理方案（适合低承载服务器） #

如果服务器不支持并行请求，用concatMap按顺序执行，并发数默认1：

from(studentList).pipe(
  concatMap(student => createTestTask(student)),
  toArray(),
  tap(allResults => {
    // 统计逻辑同上
  })
).subscribe(/* 回调逻辑同上 */);

原分组forkJoin的适配方案 #

如果你偏好之前的分组思路，可结合bufferCount和forkJoin控制组级并发：

from(studentList).pipe(
  bufferCount(20), // 每20个学生分为一组
  // 同时处理2组，每组内用forkJoin并行
  mergeMap(group => forkJoin(group.map(student => createTestTask(student))), 2),
  // 展开每组的结果数组
  concatAll(),
  toArray()
).subscribe(/* 回调逻辑同上 */);

关键注意事项 #

• 捕获单个请求错误：必须在单个API调用中捕获错误，否则某一个请求失败会导致整个流终止，无法收集后续结果。
• 合理设置并发数：根据服务器的QPS限制调整，避免因并发过高触发服务器报错。
• 进度跟踪（可选）：若需要显示进度，可在mergeMap内添加tap操作，每完成一个请求更新进度条。

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