如何在RxJS（或通用FRP）中合理组织异步响应？

要实现只保留按时序到达的有效异步响应（即仅输出那些序号大于已输出结果最大序号的响应），可以通过给每个请求分配递增序号，再结合过滤逻辑来实现。具体步骤如下：

实现代码 #

// 1. 为source$的每个事件添加递增序号
const indexedSource$ = source$.pipe(
  scan((acc, value) => ({ index: acc.index + 1, value }), { index: 0 })
);

// 2. 发起异步操作并保留序号
const asyncWithIndex$ = indexedSource$.pipe(
  mergeMap(({ index, value }) => 
    someAsyncOperation(value).pipe(
      map(result => ({ index, result }))
    )
  )
);

// 3. 过滤仅保留序号大于已输出最大序号的结果
const result$ = asyncWithIndex$.pipe(
  scan((state, { index, result }) => {
    if (index > state.maxIndex) {
      return { maxIndex: index, output: result };
    }
    return { ...state, output: null };
  }, { maxIndex: 0, output: null }),
  filter(state => state.output !== null),
  map(state => state.output)
);

也可以封装成自定义操作符，方便复用：

function orderedMergeMap(project) {
  return source$ => source$.pipe(
    scan((acc, value) => ({ index: acc.index + 1, value }), { index: 0 }),
    mergeMap(({ index, value }) => project(value).pipe(map(result => ({ index, result })))),
    scan((state, { index, result }) => ({
      maxIndex: index > state.maxIndex ? index : state.maxIndex,
      output: index > state.maxIndex ? result : null
    }), { maxIndex: 0, output: null }),
    filter(state => state.output !== null),
    map(state => state.output)
  );
}

// 使用方式
const result$ = source$.pipe(
  orderedMergeMap(s => someAsyncOperation(s))
);

逻辑说明 #

1. 序号标记：通过scan为每个source事件分配递增的序号，确保每个请求的顺序可追踪。
2. 异步绑定：用mergeMap发起异步操作，同时将序号与结果绑定，保留请求的顺序信息。
3. 过滤有效结果：再次用scan跟踪已输出结果的最大序号，仅当当前结果的序号大于该最大值时，才将其标记为可输出；最后过滤掉无效结果，得到符合预期的输出流。

对应场景验证 #

针对你提到的轮询场景：

source: -----1-----2----3------->
result1:     --------|
result2:           -----------|
result3:                ----|

• result1的序号为1，大于初始maxIndex(0)，输出1，maxIndex更新为1。
• result3的序号为3，大于当前maxIndex(1)，输出3，maxIndex更新为3。
• result2的序号为2，小于当前maxIndex(3)，被过滤丢弃。
  最终输出与预期一致：-----------1------3--->

针对RTT大于轮询间隔的场景：

source: -----1----2----3----4----5----->
result1:     --------|
result2:          --------|
result3:               --|
result4:                    -------|
result5:                         ----|

• result1（序号1）输出，maxIndex=1。
• result3（序号3）输出，maxIndex=3。
• result2（序号2）被过滤。
• result4（序号4）输出，maxIndex=4。
• result5（序号5）输出，maxIndex=5。
  最终输出与预期一致：-----------1---3---------4-5->

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