如何修改RxJS管道合并连续update事件以减少HTTP请求

要实现将连续的update事件合并，同时保留事件顺序并确保HTTP请求按序发送，可以通过RxJS的scan、endWith、mergeMap和concatMap操作符组合实现，具体步骤如下：

核心思路 #

1. 用scan操作符累积连续的update事件，遇到非update事件时，先发射合并后的update事件，再发射当前的非update事件。
2. 用endWith(null)标记流的结束，确保最后一批未处理的update事件能被合并发射。
3. 最后通过concatMap保持HTTP请求的顺序执行，和原逻辑一致。

修改后的完整代码 #

import { from, Observable, EMPTY } from 'rxjs';
import { concatMap, scan, mergeMap, endWith } from 'rxjs/operators';

type Update = number[];

interface Event {
  type: 'add' | 'delete' | 'update';
  data: Update;
}

const eventStream = from([
  { type: 'update', data: [1] },
  { type: 'update', data: [2] },
  { type: 'add', data: [3] },
  { type: 'update', data: [4] },
  { type: 'update', data: [5] },
  { type: 'delete', data: [6] },
  { type: 'update', data: [7] },
  // ... other events
]);

function postEvent(event: Event): Observable<any> {
  // ... post event to server
}

/*
将多个连续的update事件合并为单个update事件:
 { type: 'update', data: [1] }, { type: 'update', data: [2] }, => { type: 'update', data: [1, 2] }
*/
function combineUpdates(updates: Event[]): Event {
  return { type: 'update', data: updates.map(e => e.data).flat()};
}

// 修改后的管道逻辑
eventStream.pipe(
  // 添加流结束标记，确保最后一批update被处理
  endWith(null as Event | null),
  // 累积连续的update事件，生成待发射的事件批次
  scan((acc, event) => {
    if (event === null) {
      // 流结束，处理剩余的update事件
      const emit: Event[] = [];
      if (acc.updates.length > 0) {
        emit.push(combineUpdates(acc.updates));
      }
      return { updates: [], emit };
    } else if (event.type === 'update') {
      // 累积update事件，暂不发射
      return { updates: [...acc.updates, event], emit: null };
    } else {
      // 遇到非update事件，先发射累积的update（如果有），再发射当前事件
      const emit: Event[] = [];
      if (acc.updates.length > 0) {
        emit.push(combineUpdates(acc.updates));
      }
      emit.push(event);
      return { updates: [], emit };
    }
  }, { updates: [] as Event[], emit: null as Event[] | null }),
  // 将待发射的事件批次展开为单个事件流
  mergeMap(state => state.emit ? from(state.emit) : EMPTY),
  // 保持原逻辑：按序发送HTTP请求
  concatMap(event => postEvent(event))
).subscribe();

代码说明 #

• endWith(null)：在事件流结束时插入一个null标记，触发最后一批累积的update事件的合并与发射。
• scan操作符：维护一个累积状态，包含当前未发射的update数组和待发射的事件列表。当遇到非update事件或流结束时，生成待发射的事件批次。
• mergeMap：将scan输出的事件批次展开为单个事件流，确保每个合并后的update和非update事件都能被正确处理。
• concatMap：保持原有的顺序执行逻辑，确保HTTP请求按事件顺序依次发送，避免并发问题。

效果验证 #

原eventStream会被转换为以下postEvent调用顺序，完全符合需求：

{ type: 'update', data: [1, 2] },
{ type: 'add', data: [3] },
{ type: 'update', data: [4, 5] },
{ type: 'delete', data: [6] },
{ type: 'update', data: [7] },

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