使用 RxJs 实现队列系统答案

【问题标题】：implement queue system with RxJs使用 RxJs 实现队列系统
【发布时间】：2018-05-03 08:01:05
【问题描述】：

我们已经用 Javascript 实现了一个队列系统。

消费者记录项目并在内部将其插入队列（这是存储在会话存储中的数组）。

时间间隔用于出列一些项目并将这些项目发送到后端。

我们如何使用 Rxjs 和流来实现这一点？

【问题讨论】：

【解决方案1】：

const queue = new Rx.Subject();

const queueProcessing = queue
  .mergeMap(i => Rx.Observable.of(i)
              .do(val => console.log('processing item: ' + val))
              .delay(2000) /* stub processing time*/
           , 2) /* concurrency */
  .subscribe();

queue.next('asdf');
queue.next('as');
queue.next('zxvc');
queue.next('`1`');
queue.next('zx');
queue.next('234');
queue.next('5');
queue.next('645');
queue.next('asdf');
queue.next('3');
queue.next('2');
queue.next('34');
queue.next('asdf');
queue.next('5');

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

.mergeMap 包含一个演示“处理”函数，它只是在延迟后发出值。每次您想将项目添加到队列中时，您 .next() 将其添加到主题中。

【讨论】：

队列本身应该是会话存储的包装器（项目放在会话存储中）。然后它应该每 2 秒从该队列中取出 10 个项目。
为什么要求将它们放入会话存储中？您想保留请求队列吗？因为与流数据结构混合的非流数据结构（在会话存储、数组中）可以工作，但会很快变得复杂且容易出错
嗨，马克，说我想要一个队列，其中任何时候只异步处理 3 个项目，我该怎么做？这就是我的意思：如果我有一组项目要上传到后端并且我不想 1. 在下一个之前异步/等待每个 2. 发送所有内容 - 这可能导致写入热点或速率限制 3. 做一个 promise.race，因为这最终会导致 (2) 4. 做一个 promise.all，因为如果有一个长时间运行的上传，那么这个过程会变得很慢。我想要做的是： 1. 有一个所有上传的队列，一次有 3 个项目的堆栈
2.当一个项目离开堆栈时，我们将一个新项目添加到队列 3。在任何一点，堆栈中总是有 3 个项目。为了扩展您的解决方案，我将如何解决这个问题？
@ChukwumaNwaugha 检查 mergeMap 的并发参数，在我的示例中设置为 2 但可以设置为 n

【解决方案2】：

你要找的接线员是bufferTime

import { Subject } from 'rxjs/Subject';
import { bufferTime } from 'rxjs/operators';

const queue$ = new Subject();
const interval = 2000; // 2 seconds

queue$.pipe(bufferTime(interval))
    .subscribe(queueArray => {
        // Send to backend
        console.log(queueArray);
    });

queue$.next('hello');
queue$.next('world');

// After 2 seconds logs: ['hello', 'world']

【讨论】：