RxJS 如何实现保留所有触发值的leading、trailing节流缓冲操作符

• 输出错位的核心是publish多播场景下的订阅注册顺序问题，和throttleTime的参数配置无关：
  • 原代码在publish的回调中，先声明并订阅了throttleCalls$节流分支，再把共享源连接到buffer的值收集端
  • 内部共享的Subject会严格按照订阅注册顺序分发上游值：先把值推给throttleTime操作符，触发leading模式的即时切分信号
  • 但此时buffer的收集端还没来得及把当前值存入缓冲区，切分动作拿到的是空集合，就会输出空数组
  • 等值被推送到buffer端存入缓冲区后，trailing模式在窗口结束时触发切分，就会把之前漏掉的所有值打包进来，最终造成整体时序错位一格，出现空数组、值偏移的问题。

只需要调整订阅注册顺序：让buffer的值收集端优先拿到上游值存入缓冲区，再让节流分支处理值、发送切分信号，不需要额外的过滤、延时等兼容逻辑。
可直接运行的完整代码：

import { pipe, interval, asyncScheduler } from "rxjs";
import { publish, throttleTime, buffer, take } from "rxjs/operators";

const throttledBuffer = (throttleTimeMs) => {
  return pipe(
    // 多播共享上游，避免重复执行副作用
    publish((shared$) => {
      // 先定义buffer收集管道，让收集端优先注册为共享源的订阅者
      return shared$.pipe(
        buffer(
          // 再注册节流信号分支，保证值先入缓冲区，再收到切分指令
          shared$.pipe(
            throttleTime(throttleTimeMs, asyncScheduler, {
              leading: true,
              trailing: true
            })
          )
        )
      );
    })
  );
};

// 测试用例：每50ms发一个值，节流窗口100ms
const clicks$ = interval(50);
clicks$.pipe(throttledBuffer(100), take(4)).subscribe(console.log);

修正后的输出和预期完全匹配：

[0]
[1, 2]
[3]
[4, 5]

这个实现完全覆盖需求场景：

• 空闲期后第一次值到达时，立刻将当前值打包为数组发射（对应leading语义）
• 从首次值到达开始算，一个节流窗口内的所有后续值，会在窗口结束时打包为数组发射（对应trailing语义）
• 所有被节流跳过的值都会被完整保留在结果数组中，不会丢弃

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