使用 RxJS 实现去抖动批处理队列答案

【问题标题】：Implementing debouncing batching queue with RxJS使用 RxJS 实现去抖动批处理队列
【发布时间】：2017-12-23 23:55:25
【问题描述】：

我试图了解 RxJS 是否适合解决此节点模块执行的问题：https://github.com/ericdolson/debouncing-batch-queue

它的描述说：“当达到其大小或超时时，将发出并清除其内容的队列。非常适合为批量 API 聚合数据，其中及时批处理是最好的。或者任何真正需要批处理数据的地方。 "

如果是这样，有人可以指导我如何使用 RxJS 实现这个 npm 模块中的简单示例吗？如果可能的话，最好使用 ES5。

【问题讨论】：

标签： rxjs

【解决方案1】：

有一个操作符^™：bufferwithtimeorcount。如果您需要它真正等效，则输入流将是Subject，group_by 用于命名空间，如下所示：

var dbbq$ = new Subject();
dbbq$.group_by(function(v_ns) { return v_ns[1]; })
     .flatMap(function(S) {
       return S.bufferwithtimeorcount(1000, 2)
     });
dbbq$.next([ 'ribs 0' ]);
dbbq$.next([ 'more ribs', 'bbq1' ]);

// is analogous to

var dbbq = new DBBQ(1000, 2);

dbbq.add('ribs 0');
dbbq.add('more ribs', 'bbq1');

【讨论】：

从链接中可以看出，bufferwithtimeorcount 是 RxJS v 4 版本，遗憾的是 RxJS v 5 中没有。 相反，使用'bufferTime'和第三个参数来设置计数。 Ref

【解决方案2】：

我不可能用 ES5 做这个:)

const dataWithNamespace = (data, namespace) => ({data, namespace});

const source = [
  dataWithNamespace('ribs 0'),
  dataWithNamespace('ribs 1'),
  dataWithNamespace('ribs 2'),
  dataWithNamespace('ribs 3'),
  dataWithNamespace('ribs 4'),
  dataWithNamespace('more ribs', 'bbq1'),
  dataWithNamespace('more ribs', 'bbq1'),
  dataWithNamespace('brisket', 'best bbq namespace')
];

const DBBQ = (debounceTimeout, maxBatchSize) => 
  source$ => source$
    .groupBy(x => x.namespace)
    .mergeMap(grouped$ => grouped$
      .switchMap(x => 
        Rx.Observable.of(x.data)
          .concat(Rx.Observable.of(undefined)
            .delay(debounceTimeout)
          )
      )
      .bufferCount(maxBatchSize)
      .filter(x => x.length == maxBatchSize)
      .map(x => x.filter(x => x !== undefined))
    );

const source$ = Rx.Observable.from(source);
  
DBBQ(1000, 2)(source$).subscribe(console.log)

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"&gt;&lt;/script&gt;

【讨论】：