在源发出值后，Rxjs 在指定时间内缓冲发出的值答案

【问题标题】：Rxjs buffer the emitted values for specified time after source emitted values在源发出值后，Rxjs 在指定时间内缓冲发出的值
【发布时间】：2021-09-30 04:09:40
【问题描述】：

如果有一些事件触发，我有一个 source$ observable 收集数据流。我想将这些在指定时间发生的数据收集到数组中。

const eventSubject = new Subject();
eventSubject.next(data); 

const source$ = eventSubject.asObservable();
source$.pipe(takeUntil(destroyed$)).subscribe(
    data => {
      console.log(data);
    }
);

上面的 source$ 立即处理发出的数据。

现在我想改进这一点，等待几秒钟并收集在指定时间发生的所有数据并发出一次。所以我修改为与 bufferTime 一起使用，如下所示：

const source$ = eventSubject.asObservable();
source$.pipe(takeUntil(destroyed$), bufferTime(2000)).subscribe(
    data => {
      console.log(data);
    }
);

在使用 bufferTime 进行测试后，我发现它每 2 秒发出一次，即使源没有接收数据。如果源没有接收到数据，它会发出空对象。

我想要的只是当 source$ 接收数据，然后开始缓冲 2s，然后发出值。如果 source$ 没有接收到数据，它不应该发出任何东西。

我检查了 bufferWhen、windowWhen、windowTime 并不都符合我的要求。它们在每个指定的时间间隔发射。

有没有其他操作员可以做我想做的事？

非常感谢。

【问题讨论】：

标签： rxjs

【解决方案1】：

您可以添加一个filter 运算符来忽略空对象发射

const source$ = eventSubject.asObservable();
source$.pipe(takeUntil(destroyed$), bufferTime(2000),filter(arr=>arr.length)).subscribe(
    data => {
      console.log(data);
    }
);

【讨论】：

还有一个定时器正在运行。在我看来，每 2 秒计时器都很昂贵且浪费资源。所以我不想在没有数据开始接收时触发计时器。
一旦它发出一个空对象，您就可以完成流来完成。但在这种情况下，流已经死了，不再在你的监视之下。

【解决方案2】：

我会选择connect(shared$ => ...) 和buffer(signal$)。

我认为是这样的：

source$.pipe(
  connect(shared$ => shared$.pipe(
    buffer(shared$.pipe(
      debounceTime(2000)
    ))
  ))
)

connect 创建一个共享的 observable，这样您就可以在源上拥有多个订阅，而无需实际打开这些订阅。

在那里我运行了一个buffer，它的选择器是同一个源的debounceTime，这样它就可以去抖动那么多（即当source$不发出超过2秒时会发出数组）

也许你需要的是throttleTime(2000, { leading: false, trailing: true }) 而不是 debounceTime。这取决于您的用例。

【讨论】：

我稍后会测试这是否符合我的要求。非常感谢。

【解决方案3】：

这种情况的最佳解决方案是使用带有 debouceTime 运算符通知器的缓冲区运算符。

例如：

const source$ = eventSubject.asObservable();
source$
  .pipe(
  // buffer: accumulate emitions to an array,until closing notifier emits. (closing notifier is the argument below)
    buffer( 
      // debounceTime : will grab the emit that afterwards 2 seconds has passed without another emit.
      $source.pipe(debounceTime(2000))
).subscribe(
   // will return the data that has been emitted througout the 2 seconds in a form of an array , where each item in the array is the emits, by the order they were triggered.
    data => {
      console.log(data); 
    }
);

buffer:

缓冲源 Observable 值，直到 closeNotifier 发出。

debounceTime:

仅在经过特定时间跨度且没有其他源发射时才从源 Observable 发出通知。

与过滤运算符解决方案相比，此解决方案不会使间隔/计时器保持活动状态。而且它非常干净优雅的 IMO。

【讨论】：