RxJS中两个mergeMap未实现惰性执行的问题咨询

问题原因 #

• Promise异步特性打乱时序：你使用Promise.resolve()返回的Promise会被推入微任务队列，无法同步发射值。当源of(1,2,...)同步发射第一个值1时，mergeMap调用映射器得到Promise，但对应的Observable要等到当前同步代码执行完才会发射数组[1,1]。
• 下游需求提前触发拉取：take(3)需要3个值，当下游第一次请求值却没收到响应时，会继续请求更多值，导致mergeMap提前从源拉取下一个值2，触发不必要的映射调用。
• concatAll展开时机滞后：Promise resolve后concatAll才会展开数组，但此时mergeMap已经因为下游的需求提前拉取了源的后续值，造成映射器多次执行。

解决办法 #

核心思路是让内部Observable能同步反馈值的产生，让RxJS的拉取机制正确控制源的发射节奏，避免提前触发映射器。

方案1：用of替代Promise.resolve（推荐） #

直接返回Observable而非Promise，让内部值同步发射，mergeMap能准确感知值的产生，从而只在需要时从源拉取新值：

import { of, mergeMap, concatAll, toArray, take } from 'rxjs'

const duplicateValue = log => value => {
  console.log(log, value)
  return of([value, value]); // 替换Promise.resolve为of
}

of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).pipe(
  mergeMap(duplicateValue('mapping a'), 1),
  concatAll(),
  mergeMap(duplicateValue('mapping b'), 1),
  concatAll(),
  take(3),
  toArray(),
)
.subscribe(result => {
  console.log(result);
});

执行输出：

mapping a 1
mapping b 1
mapping b 1
[1, 1, 1]

完全符合预期：mapping a仅运行1次，mapping b运行2次。

方案2：若必须用Promise，用from包裹 #

如果业务场景中依赖Promise，用from将其转为Observable，确保RxJS能正确处理内部值的发射：

import { of, mergeMap, concatAll, toArray, take, from } from 'rxjs'

const duplicateValue = log => value => {
  console.log(log, value)
  return from(Promise.resolve([value, value])); // 用from包裹Promise
}

// 管道代码保持不变

此方案效果和方案1一致，但Promise的异步特性仍会存在微小时序差异，优先推荐方案1。

方案3：用concatMap简化代码 #

concatMap本质是并发数为1的mergeMap，可以省略并发数参数，代码更简洁：

import { of, concatMap, concatAll, toArray, take } from 'rxjs'

const duplicateValue = log => value => {
  console.log(log, value)
  return of([value, value]);
}

of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).pipe(
  concatMap(duplicateValue('mapping a')),
  concatAll(),
  concatMap(duplicateValue('mapping b')),
  concatAll(),
  take(3),
  toArray(),
)
.subscribe(result => {
  console.log(result);
});

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