如何在 RXJS 中的缓冲区后累积变化答案

【问题标题】：how to accumulate change after buffer in RXJS如何在 RXJS 中的缓冲区后累积变化
【发布时间】：2022-08-02 21:41:28
【问题描述】：

在 RXJS 的世界中，我有一个我想缓冲/批处理的 observables 流。为此，我使用可爱的 bufferCount(5) 来减慢速度

bufferCount(5)

然后我想解决一个异步调用（保存到数据库），这是缓冲的 observable。

bufferCount(5),
switchMap(b => forkJoin(b)),
// unbuffer here

然后，我想取消缓冲或累积更改以获取单个更改列表。为此，我使用了bufferTime(2000) 并用地图跟踪它以使结果变平......不用说，我觉得这样做很肮脏。

如何在不使用计时器的情况下取消缓冲流？

你能解释一下真实的用例吗？因为我不明白你试图用 buffer + forkJoin 组合来实现什么，更不用说使用 switchMap 运算符可能取消以前的批次了。干杯
如果您想获得所有更改的单个列表，您可以使用scan 并将当前更改合并到一个大数组中。
添加了代码沙箱。

标签： javascript typescript rxjs

【解决方案1】：

听起来您在问如何将对象列表/可观察对象转换为每个对象的 DB-save-result 值的可观察对象，同时避免一次保存每个对象。

在撰写本文时，您的代码框似乎在使用mergeAll with a specified concurrency value 的正确轨道上。

这将是我的建议：

const perform$ = rick$.pipe(
  switchMap((response) => response.json()),
  concatMap((x) => {
    return x.results.map((r) =>
      longRunningTask(r, Math.floor(Math.random() * 5000))
    );
  }),
  mergeAll(5)
);

您可以创建一个任务流（订阅时启动保存操作的可观察对象），而不是关心自己的缓冲和“取消缓冲”，并使用mergeAll(N) 一次运行 N 个任务。每个任务结果都将包含在这个 observable 中。

我已经包含了一个更新的代码框。

【讨论】：