JS并发数控制

doWork-concurrent-limit.js

// 实现 Promise.allConcurrent
// Promise.allConcurrent(2)([()=>fetch('BLAH1'), ()=>fetch('BLAH2'),...()=>fetch('BLAHN')])
// 先同时执行 BLAH1、BLAH2，BLAH1 执行完毕之后立即开始执行 BLAH3，以此类推……
Promise.allConcurrent = n => list => promiseAllStepN(n, list)

/**
 * 带并发数限制的 Promise.All
 * @param {*} n 同时并发数
 * @param {*} list 需要执行的 Promise 列表
 */
function promiseAllStepN(n, list) {
    // 去掉前 n 个之后的列表
    const tail = list.splice(n)
    const head = list
    const resolved = []
    let processed = 0

    return new Promise(resolve => {
        head.forEach(x => {
            // 这里创建 Promise，开始运行
            const res = x()
            // 放入结果
            resolved.push(res)
            res.then(y => {
                // 一个 Promise 执行完成之后，继续执行新的 Promise
                runNext()
                return y
            })
        })

        function runNext() {
            if (processed === tail.length) {
                // 等待全部完成之后整体 resolve
                resolve(Promise.all(resolved))
            } else {
                // 这里创建下一个 Promise，并且开始运行
                // 注意 resolved.push 和 return x 对应
                resolved.push(tail[processed]().then(x => {
                    // 一个 Promise 执行完成之后，继续执行新的 Promise
                    runNext()
                    return x
                }))
                processed++
            }
        }/*  */
    })
}

let doWork = (workid) => {
    return new Promise(resolve => {
        console.log('start running', workid)
        // 模拟大量任务
        setTimeout(() => {
            console.log('finished running', workid);
            resolve(workid);
        }, 10000 * Math.random());
    });
};

works = []
for (let i = 0; i < 30; ++i) {
    // 创建新的函数而不是直接创建 Promise
    works.push(() => doWork(i))
}
Promise.allConcurrent(3)(works)