RxJS如何让Observable两次发射值的时间间隔不低于1000毫秒

实现方案 #

你可以通过RxJS的concatMap操作符实现该需求，该方案不会丢失observableA的任何发射值，同时仅在两次发射间隔不足1000ms时补全等待时间，原本就超过1000ms的间隔会保持不变，完全匹配你给出的示例效果。

核心代码

import { concatMap, of, timer, ignoreElements, concat } from 'rxjs';

const observableB = observableA.pipe(
  concatMap(value => {
    // 先立即发射当前值，再等待1000ms后结束当前内部流，等待期间无任何输出
    return concat(
      of(value),
      timer(1000).pipe(ignoreElements())
    )
  })
);

逻辑说明

• concatMap会严格按照observableA的发射顺序处理值，必须等前一个值对应的内部流完全结束，才会处理下一个值
• 每个值对应的内部流分两步：第一步立即发射当前值，第二步启动1000ms计时器，计时器结束后内部流才结束
• 如果observableA两个相邻值的发射间隔小于1000ms：后一个值会等前一个值的1000ms等待计时结束后才会发射，自动补全间隔到1000ms
• 如果observableA两个相邻值的发射间隔大于等于1000ms：前一个值的内部流早就完成等待，后一个值到达后会立即发射，间隔保持原长度

内容的提问来源于stack exchange，提问作者code đờ