利用 JavaScript 实现并发控制的示例代码

const task = timeout => new Promise((resolve) => setTimeout(() => {
  resolve(timeout);
 }, timeout))

 const taskList = [1000, 3000, 200, 1300, 800, 2000];

 async function startNoConcurrentControl() {
  console.time(NO_CONCURRENT_CONTROL_LOG);
  await Promise.all(taskList.map(item => task(item)));
  console.timeEnd(NO_CONCURRENT_CONTROL_LOG);
 }

 startNoConcurrentControl();

class Queue {
  constructor() {
   this._queue = [];
  }

  push(value) {
   return this._queue.push(value);
  }

  shift() {
   return this._queue.shift();
  }

  isEmpty() {
   return this._queue.length === 0;
  }
 }

class DelayedTask {
  constructor(resolve, fn, args) {
   this.resolve = resolve;
   this.fn = fn;
   this.args = args;
  }
 }

class TaskPool {
  constructor(size) {
   this.size = size;
   this.queue = new Queue();
  }

  addTask(fn, args) {
   return new Promise((resolve) => {
    this.queue.push(new DelayedTask(resolve, fn, args));
    if (this.size) {
     this.size--;
     const { resolve: taskResole, fn, args } = this.queue.shift();
     taskResole(this.runTask(fn, args));
    }
   })
  }

  pullTask() {
   if (this.queue.isEmpty()) {
    return;
   }

   if (this.size === 0) {
    return;
   }

   this.size++;
   const { resolve, fn, args } = this.queue.shift();
   resolve(this.runTask(fn, args));
  }

  runTask(fn, args) {
   const result = Promise.resolve(fn(...args));

   result.then(() => {
    this.size--;
    this.pullTask();
   }).catch(() => {
    this.size--;
    this.pullTask();
   })

   return result;
  }
 }

 const cc = new ConcurrentControl(2);

 async function startConcurrentControl() {
  console.time(CONCURRENT_CONTROL_LOG);
  await Promise.all(taskList.map(item => cc.addTask(task, [item])))
  console.timeEnd(CONCURRENT_CONTROL_LOG);
 }

 startConcurrentControl();

await Promise.all(taskList.map(item => cc.addTask(task, [item])))

addTask(fn) {
  return (...args) => {
   return new Promise((resolve) => {
    this.queue.push(new DelayedTask(resolve, fn, args));

    if (this.size) {
     this.size--;
     const { resolve: taskResole, fn: taskFn, args: taskArgs } = this.queue.shift();
     taskResole(this.runTask(taskFn, taskArgs));
    }
   })
  }
 }

 await Promise.all(taskList.map(cc.addTask(task)))

const Benchmark = require('benchmark');
 const suite = new Benchmark.Suite;

 suite.add('shift', function() {
  let count = 10;
  const arr = generateArray(count);
  while (count--) {
   arr.shift();
  }
 })
 .add('reverse + pop', function() {
  let count = 10;
  const arr = generateArray(count);
  arr.reverse();
  while (count--) {
   arr.pop();
  }
 })
 .on('cycle', function(event) {
  console.log(String(event.target));
 })
 .on('complete', function() {
  console.log('Fastest is ' + this.filter('fastest').map('name'));
  console.log('\n')
 })
 .run({
  async: true
 })

 class HighPerformanceQueue {
  constructor() {
   this.q1 = []; // 用于 push 数据
   this.q2 = []; // 用于 shift 数据
  }

  push(value) {
   return this.q1.push(value);
  }

  shift() {
   let q2 = this.q2;
   if (q2.length === 0) {
    const q1 = this.q1;
    if (q1.length === 0) {
     return;
    }
    q2 = this.q2 = q1.reverse();
   }

   return q2.pop();
  }

  isEmpty() {
   if (this.q1.length === 0 && this.q2.length === 0) {
    return true;
   }
   return false;
  }
 }