rxjs 速率限制（每秒请求数）和并发

Question

我正在尝试弄清楚如何在 rxjs 中编写速率限制器。用于访问大多数 api（twitter、facebook 等）如果不支持开箱即用的方法，我会假设可以编写调度程序。例如 highland.js 有ratelimit。我不想丢弃任何项目，如窗口、样本等。

var source = Rx.Observable.create(function (observer) {

  // queue of requests
  _.each(requests, function(r) {
    observer.onNext(r);
  });

  observer.onCompleted();

  // Any cleanup logic might go here
  return function () {
    console.log('disposed');
  }
})
  // what goes here, if built in (e.g. 2 requests per 2 seconds or 15 request per 15 minutes)

// SHOULD ONLY RUN
var subscription = source.subscribe(
  function (x) { console.log('onNext: %s', x); },
  function (e) { console.log('onError: %s', e); },
  function () { console.log('onCompleted'); });

编辑 1： 想想这样的事情，使用令牌桶算法，还是很粗糙但是……

Rx.Observable.prototype.tokenBucket = function(options, scheduler) {
  function time() {
    return new Date().getTime();
  }

  var BUCKET = {
    capacity: options.capacity || Infinity,
    left: options.capacity,
    last: time(),

    tokensPerInterval: options.tokensPerInterval,
    interval: options.interval
  };

  //var BUCKET = _.merge(defaultOptions, options);
  console.log(BUCKET);

  var source = this,
    scheduler = scheduler || (scheduler = Rx.Scheduler.timeout);

  return Rx.Observable.create(function(observer) {
    var d1 = source.subscribe(function(mainValue) {
      return throttle(mainValue);
    });

    function throttle(x, tokens) {
      if (BUCKET.capacity === Infinity) {
        return observer.onNext(x);
      } // return x;

      // the number of tokens to add every S milliseconds = (r*S)/1000.
      var self = BUCKET;
      var now = time();

      var deltaMS = Math.max(now - self.last, 0);
      self.last = now;
      var dripAmount = deltaMS * (self.tokensPerInterval / self.interval);
      self.left = Math.min(self.left + dripAmount, self.capacity);

      if (self.left < 1) {
        var interval = Math.ceil((1 - self.left) * self.interval);
        scheduler.scheduleWithRelative(interval, function (s, i) {
          return throttle(x);
        });
      } else {
        self.left -= tokens || 1;
        console.log('calling');
        return observer.onNext(x);
      }
    }

    return function() {
      d1.dispose();
      console.log('disposed tokenBucket');
    };
  });
};

var start = moment();
var source = Rx.Observable.range(1, 20)
  .tokenBucket({capacity: 2, tokensPerInterval: 2, interval: 2000})

var subscription = source.subscribe(
  function (x) { console.log('onNext: %s', x); addToDom(x); },
  function (e) { console.log('onError: %s', e); },
  function () { console.log('onCompleted'); });

function addToDom(x) {
  var ul = document.getElementById('c');
  var li = document.createElement('li');
  li.innerHTML = x + ' - ' + moment().diff(start, 'seconds') + 's ago';
  ul.appendChild(li);
}

<script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.10.3/moment.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/2.5.3/rx.all.js"></script>
<ul id="c"></ul>

Answer 1

如果您只想删除发生在两者之间的事件，可以使用 windowWithTimeOrCount + throttleFirst：

var subscription = source

 //Splits the events into 15 minute windows
 .windowWithTimeOrCount(900000 /*15 minutes*/, 15) 

 //Stops us from receiving more than one window in 15 minutes
 .throttleFirst(900000 /*15 minutes*/)

 //Flatten the observable
 .concatAll()
 .subscribe(
  function (x) { console.log('onNext: %s', x); },
  function (e) { console.log('onError: %s', e); },
  function () { console.log('onCompleted'); });

工作示例（控制台输出）：

var source = Rx.Observable.generateWithRelativeTime(
    0,
    function(x) { return x < 1000; },
    function(x) { return x + 1; },
    function(x) { return x; },
    function(x) { return Math.floor(Math.random() * 100); });


source
 .windowWithTimeOrCount(1000, 3)
 .throttleFirst(1000)
 .concatAll()
 .subscribe(console.log.bind(console));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/2.5.3/rx.all.js"></script>

备选方案 1

如果您不想删除任何值，也可以在管道上使用 controlled 以及特别滚动的 regulate 方法：

var subscription = source
  .windowWithTimeOrCount(900000, 15)
  //This will stop the loss of any events from the hot window
  .map(function(x) {
    var c = x.replay(),
        d = c.connect();
    //Shut down the connected observable when you are done.
    return Rx.Observable.using(function() {return d; },
                               function() {return c; });
  })
  //This will prevent more than one window from going through per interval
  //[See snippet]
  .regulate(900000)
  .concatAll()
  .subscribe(
    function (x) { console.log('onNext: %s', x); },
    function (e) { console.log('onError: %s', e); },
    function () { console.log('onCompleted'); });

工作示例（控制台输出）：

Rx.Observable.prototype.regulate = function(interval, scheduler) {
  var source = this,
    scheduler = scheduler || (scheduler = Rx.Scheduler.timeout);

  return Rx.Observable.create(function(observer) {
    var controller = source.controlled(scheduler),
      d = new Rx.SerialDisposable();

    function nextSample(x) {

      //This will request a new value after our minimum interval expires
      d.setDisposable(scheduler.scheduleWithRelative(interval, function(s) {
        return controller.request(1);
      }));

      observer.onNext(x);
    }

    return new Rx.CompositeDisposable(
      d,
      controller.subscribe(nextSample,
        observer.onError.bind(observer),
        observer.onCompleted.bind(observer)),
      controller.request(1));

  }, source);

};


var subscription = Rx.Observable.generateWithRelativeTime(
    0,
    function(x) {
      return x < 100;
    },
    function(x) {
      return x + 1;
    },
    function(x) {
      return x;
    },
    function(x) {
      return Math.floor(Math.random() * 200);
    })
  //Divides up windows by count and our interval time
  .windowWithTimeOrCount(2000, 15)
  //Necessary since the windows we receive are hot
  .map(function(x) {
    var c = x.replay();
    var d = c.connect();
    return Rx.Observable.using(function() {
      return d;
    }, function() {
      return c;
    });
  })
  //Only allows one window through every 2 seconds
  .regulate(2000)
  //Flatten everything out
  .concatAll()
  .subscribe(console.log.bind(console), console.error.bind(console), console.log.bind(console, "Finished!"));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/2.5.3/rx.all.js"></script>

Answer 2

我在我的个人项目中遇到了一个非常相似的问题，并决定将可重用的解决方案发布为 npm 包https://www.npmjs.com/package/rx-op-lossless-throttle

与 http://www.g9labs.com/2016/03/21/lossless-rate-limiting-with-rxjs/ 不同，它不会强制延迟每个事件。

Answer 3

如果您不想丢失任何通知，可以使用缓冲区或其变体之一（使用时间/计数/时间或计数）。 它基本上将通知组合在一个数组中，并在以下情况下转发该数组：

• 一个可观察的通知
• 计时器到期
• 已达到通知计数
• 以上的组合。

因此，您可以将通知缓冲在一个数组中，并且每分钟只接收一次，或者在 100 个通知到达时接收它。

• buffer
• buffer with count
• buffer with time
• buffer with time or count