Node.js中如何控制指定数量的Promise并发执行？

你当前的核心问题是：Promise.race(promises)返回的是异步任务的结果值，但你的promises数组里存的是Promise实例，所以用promises.indexOf(data)根本找不到对应的Promise对象，自然无法从数组中移除已完成的任务。

下面提供两种简洁可靠的实现方案，完美实现你想要的“超市收银台”式并发控制效果：

方案一：轻量计数器式实现（推荐） #

用计数器跟踪活跃任务数，任务完成后自动启动下一个，逻辑直观且性能最优：

const buffer = [];
const MAX_CONCURRENCY = 3; // 可按需改为5
let activeTaskCount = 0;
let currentEntityIndex = 0;
const entities = [/* 你的实体列表 */];

async function startScraping() {
    // 先启动第一批任务，填满并发池
    while (activeTaskCount < MAX_CONCURRENCY && currentEntityIndex < entities.length) {
        launchNextTask();
    }
}

async function launchNextTask() {
    if (currentEntityIndex >= entities.length) return;
    
    const entity = entities[currentEntityIndex++];
    activeTaskCount++;
    
    try {
        const data = await scrapingProcessForAnEntity(entity);
        buffer.push(data);
        // 这里无需担心并发写入：JS是单线程，await后的代码在微任务队列执行，不会同时触发buffer.push
    } catch (error) {
        // 按需处理错误，比如记录日志
        console.error(`处理实体${entity}失败`, error);
    } finally {
        activeTaskCount--;
        // 当前任务结束，立刻启动下一个
        launchNextTask();
    }
}

逻辑说明 #

1. activeTaskCount记录当前正在运行的异步任务数量，currentEntityIndex标记下一个待处理的实体位置
2. 初始化时先启动MAX_CONCURRENCY个任务，填满并发池
3. 每个任务完成（无论成功/失败），都会在finally块中减少计数器，并立即启动下一个任务，直到所有实体处理完毕

方案二：Set集合跟踪活跃Promise #

如果你需要显式跟踪所有活跃任务，可使用Set存储Promise实例，完成后自动移除：

const buffer = [];
const MAX_CONCURRENCY = 3;
const entities = [/* 你的实体列表 */];

async function startScraping() {
    const taskQueue = [...entities];
    const activePromises = new Set();

    while (taskQueue.length > 0) {
        // 填满并发池
        while (activePromises.size < MAX_CONCURRENCY && taskQueue.length > 0) {
            const entity = taskQueue.shift();
            const taskPromise = scrapingProcessForAnEntity(entity)
                .then(data => buffer.push(data))
                .catch(err => console.error(`处理实体${entity}失败`, err))
                .finally(() => activePromises.delete(taskPromise));
            
            activePromises.add(taskPromise);
        }
        // 等待任意一个任务完成，再继续填充并发池
        await Promise.race(activePromises);
    }
    // 等待最后一批任务全部完成
    await Promise.all(activePromises);
}

逻辑说明 #

1. 用Set存储正在运行的Promise实例（Set支持直接删除指定对象）
2. 每次循环将并发池填满至MAX_CONCURRENCY，然后等待任意一个任务完成
3. 任务完成后在finally中从Set移除自身，循环继续填充新任务
4. 最后用Promise.all确保所有剩余任务执行完毕

内容的提问来源于stack exchange，提问作者Robson Luan