Promise并发控制 - EricYangXD/LearnReact GitHub Wiki

class Schedule{

    constructor(maxNum){
        this.list = [];
        this.maxNum = maxNum
        this.workingNum = 0
    }
    
    add(promiseCreator){
        this.list.push(promiseCreator)
    }

    start(){
        for (let index = 0; index < this.maxNum; index++) {
            this.doNext()
        }
    }

    doNext(){
        if(this.list.length && this.workingNum < this.maxNum){
            this.workingNum++;
            const promise = this.list.shift();
            promise().then(()=>{
                this.workingNum--;
                this.doNext();
            })

        }
    }
}

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

const schedule = new Schedule(2);

const addTask = (time, order)=>{
    schedule.add(()=>timeout(time).then(()=>{
        console.log(order);
    }))
}

addTask(1000, 1)
addTask(500, 2)
addTask(300, 3)
addTask(400, 4)

schedule.start()

async function asyncPool(poolLimit, array, iteratorFn) {
  const ret = []; // 存储所有的异步任务
  const executing = []; // 存储正在执行的异步任务
  for (const item of array) {
    // 调用iteratorFn函数创建异步任务
    const p = Promise.resolve().then(() => iteratorFn(item, array));
    ret.push(p); // 保存新的异步任务

    // 当poolLimit值小于或等于总任务个数时，进行并发控制
    if (poolLimit <= array.length) {
      // 当任务完成后，从正在执行的任务数组中移除已完成的任务
      const e = p.then(() => executing.splice(executing.indexOf(e), 1));
      executing.push(e); // 保存正在执行的异步任务
      if (executing.length >= poolLimit) {
        await Promise.race(executing); // 等待较快的任务执行完成
      }
    }
  }
  return Promise.all(ret);
}

function asyncPool(poolLimit, array, iteratorFn) {
  let i = 0;
  const ret = []; // 存储所有的异步任务
  const executing = []; // 存储正在执行的异步任务
  const enqueue = function () {
    if (i === array.length) {
      return Promise.resolve();
    }
    const item = array[i++]; // 获取新的任务项
    const p = Promise.resolve().then(() => iteratorFn(item, array));
    ret.push(p);

    let r = Promise.resolve();

    // 当poolLimit值小于或等于总任务个数时，进行并发控制
    if (poolLimit <= array.length) {
      // 当任务完成后，从正在执行的任务数组中移除已完成的任务
      const e = p.then(() => executing.splice(executing.indexOf(e), 1));
      executing.push(e);
      if (executing.length >= poolLimit) {
        r = Promise.race(executing); 
      }
    }
 
    // 正在执行任务列表 中较快的任务执行完成之后，才会从array数组中获取新的待办任务
    return r.then(() => enqueue());
  };
  return enqueue().then(() => Promise.all(ret));
}