RxJS v5中的速率限制和计数限制事件，但也允许传递 [英] Rate-limiting and count-limiting events in RxJS v5, but also allowing pass-through

问题描述

我有很多事件要发送到服务。但请求是速率限制的，每个请求都有一个计数限制：

I have a bunch of events to send up to a service. But the requests are rate limited and each request has a count limit:

• 每秒1个请求： bufferTime（1000 ）


• 每个请求100个事件项： bufferCount（100）

• 1 request per second: bufferTime(1000)
• 100 event items per request: bufferCount(100)

问题是，我不知道如何以一种有意义的方式组合它们。

The problem is, I am not sure how to combine them in a way that makes sense.

进一步复杂化，如果我们没有达到任何限制，我需要确保事件即时通过。

Complicating this further, I need to make sure that events go through instantaneously if we don't hit either limit.

例如，如果在非繁忙时间只有一个事件发生，我不希望它实际等待100个事件项目。

For example, I don't want it to actually wait for 100 event items before letting it go through if it's only one single event during a non-busy time.

我还发现RxJS v4中存在 bufferWithTimeOrCount ，尽管即使我拥有它，我也不确定如何使用它。

I also found that there was a bufferWithTimeOrCount that existed in RxJS v4, although I am not sure how I'd use that even if I had it.

这是我为你做的一个JSBin来测试你的解决方案：

Here is a JSBin I made for you to test your solution:

http://jsbin.com/fozexehiba/1/edit？ js，console，output

非常感谢任何帮助。

推荐答案

bufferTime（）运算符有三个参数，它们结合了 bufferTime 和<$ c的功能$ C> bufferCount 。请参见 http://reactivex.io/rxjs /class/es6/Observable.js~Observable.html#instance-method-bufferTime 。

The bufferTime() operator takes three parameters which combines the functionality of bufferTime and bufferCount. See http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-bufferTime.

使用 .bufferTime（1000， null，3）你可以每1000毫秒或当它达到3个项目时创建一个缓冲区。但是，这意味着它不能保证每个缓冲区之间有1000ms的延迟。

With .bufferTime(1000, null, 3) you can make a buffer every 1000ms or when it reaches 3 items. However, this means that it doesn't guarantee 1000ms delay between each buffer.

所以你可以使用这样的东西很容易使用（只缓冲3个项目最大1000毫秒）：

So you could use something like this which is pretty easy to use (buffers only 3 items for max 1000ms):

click$
  .scan((a, b) => a + 1, 0)
  .bufferTime(1000, null, 3)
  .filter(buffer => buffer.length > 0)
  .concatMap(buffer => Rx.Observable.of(buffer).delay(1000))
  .timestamp()
  .subscribe(console.log);

查看现场演示： http://jsbin.com/libazer/7/edit?js,console,output

您可能想要的唯一区别是第一次发射可能会延迟超过1000毫秒。这是因为 bufferTime（）和延迟（1000）运算符会延迟以确保始终至少为1000毫秒差距。

The only difference to what you probably wanted is that the first emission might be delayed by more than 1000ms. This is because both bufferTime() and delay(1000) operators make a delay to ensure that there's always at least 1000ms gap.

这篇关于RxJS v5中的速率限制和计数限制事件，但也允许传递的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！