RxJS - 拆分数组并在达到特定计数时发出答案

【问题标题】：RxJS - split the array and emit when specific count is reachedRxJS - 拆分数组并在达到特定计数时发出
【发布时间】：2023-03-22 23:03:01
【问题描述】：

我是 RxJS 的新手。我有特定的数组，我正在使用 Observable。

[a,b,c,d,e,f,g,h]

我想处理背压，每次处理 3 个元素。

我想这样拆分并发出：

[a,b,c] [d,e,f] [g,h]

我为此使用了 bufferCount(3)。但是 bufferCount 的问题在于它确实发出了最后一个间隔。因为最后一个区间只有 2 个元素。

这是我的示例代码

from(streamOfArr$).pipe(
         flatMap(somefunc()),
         bufferCount(3),
         tap((x) => {
                    console.log('3', x);
                }),
         flatMap(somefunc1()),

如何在 bufferCount 中发出最后一个间隔。

【问题讨论】：

你有没有考虑过缓冲发射应该在什么条件下发生？（即你怎么能确定在项目h 之后什么都没有？）也就是说，一种可能的解决方案是设置一个带有内部可观察对象的buffer()，它在以下情况下发射：1）收到三个项目，2）一些通过timer() 超出任意超时。
@miqh 是否有可能知道数组长度并检查数组中是否还有多少剩余项未发出，以及它们是否小于 buffermaxSize。然后 bufferMaxSize 可能会设置数组中的剩余项目或其他内容。

标签： javascript arrays rxjs vue-rx

【解决方案1】：

我相信这里还有另一个问题。

如果你看到这个例子：https://stackblitz.com/edit/rxjs-3jwznz

import { from } from 'rxjs'; 
import { bufferCount } from 'rxjs/operators';


const source = from([1,2,3,4,5,6,7,8]).pipe(
  bufferCount(3)
);

source.subscribe(x => console.log(x));

您将看到 bufferCount 发出最后一个间隔（即使最后一个只有 2 个）。你确定你的源 observable 正在完成吗？

【讨论】：

【解决方案2】：

我认为您可能希望使用并发号码检查 mergeAll 而不是 bufferCount。 mergeAll 不会将数据拆分成块，但它只会订阅下一个 observable，当另一个前一个 observable 完成时。只有 n 个 observables 同时是热的。

function obs(n: number): Observable<number> {
  return of(n).pipe(
    tap(() => console.log('Processing: ', n)),
    delay(n * 500),
  );
}

const observables$ = [obs(1), obs(2), obs(3), obs(4), obs(5), obs(6), obs(7), obs(8), obs(9), obs(10)];

from(observables$).pipe(
  mergeAll(3),
).subscribe(console.log);

另请参阅 stackblitz：https://stackblitz.com/edit/rxjs-rcruhv?file=index.ts

通过此示例，您可以在控制台中看到，开始时直接打印 3 个输出，发生订阅输出时每隔一个处理输出出现。

【讨论】：