并行“流量”控制

promise-lock.js

'use strict';

// promise lock

// const PromiseLock = (() => {
//   function PromiseLock(concurrency) {
//     this.queue = [];
//     this.lock = 0;
//     this.concurrency = Math.max(concurrency, 1);
//   }

//   PromiseLock.prototype._lock = function () {
//     return new Promise((resolve, reject) => {
//       this.lock += 1;
//       if (this.lock <= this.concurrency) resolve();
//       else this.queue.push(resolve);
//     });
//   };

//   PromiseLock.prototype._unlock = function () {
//     this.lock -= 1;
//     if (this.queue.length > 0) this.queue.shift()();
//   };

//   PromiseLock.prototype.serial = function (promisedFn) {
//     this._lock().then(promisedFn).then(this._unlock.bind(this));
//   };

//   return PromiseLock;
// })();


class PromiseLock {
  constructor(concurrency) {
    this.queue = [];
    this.lock = 0;
    this.concurrency = Math.max(1, concurrency);
  }

  _lock() {
    return new Promise((resolve, reject) => {
      this.lock += 1;
      if (this.lock <= this.concurrency) resolve();
      else this.queue.push(resolve);
    });
  }

  _unlock() {
    this.lock -= 1;
    if (this.queue.length > 0) this.queue.shift()();
  }

  serial(promisedFn) {
    this._lock().then(promisedFn).then(this._unlock.bind(this));
  }
}


//// test ////

function sleep(tick) {
  return new Promise((resolve) => {
    setTimeout(resolve, tick);
  });
}


let lock = new PromiseLock(5);
(new Array(20 + 1)).join(',').split('').forEach((e, i) => {
  lock.serial(() => {
    return sleep(1000).then(() => console.log(i)).catch(reason => console.info(reason));
  });
});

// 每秒输出5个数字，总共4秒执行完毕