愚公上岸说

11-11 23:06 四川大学前端工程师发布于四川

关注

【前端面试小册】JS-第20节：实现并发限制（滴滴、头条面试题）

第20节：实现并发限制（滴滴、头条面试题）

一、题目描述

1.1 需求分析

实现一个并发限制函数 concurrenceLimits(tasks, limit)：

参数：

tasks：任务队列（请求列表），每个任务是一个返回 Promise 的函数
limit：每次同时并发最多允许的请求数目

要求：

同时发出 n 个请求（n <= limit）
只要其中一个完成请求，立马又加入一个请求
让请求队列始终保持 n 个请求，直到没有任务再可以加入
返回的结果必须按照任务队列的顺序返回，返回结果类似于 Promise.all()

1.2 示例说明

concurrenceLimits([req1, req2, req3], 2)

执行过程：

初始阶段：同时发出请求 req1、req2（并发数为 2）
req2 完成：req1 还没有完成，因为并发最多可以有两个请求，所以这个时候 req3 也发出请求
结果返回：虽然 req2 最先返回，但是返回的结果必须按照请求队列顺序返回，即结果必须是：[res1, res2, res3]

执行流程图：

alt

二、实现思路

2.1 核心思想

任务队列管理：维护一个正在执行的任务队列
并发控制：使用 Promise.race 等待任意一个任务完成
顺序保证：使用结果数组按顺序存储结果
动态调度：任务完成后立即启动新任务

2.2 关键点

任务队列：存储正在执行的 Promise
结果队列：按顺序存储所有任务的 Promise
Promise.race：等待任意一个任务完成
动态调度：任务完成后继续执行下一个任务

三、完整实现

3.1 基础版本

function concurrenceLimits(tasks, limit) {
    let count = 0;  // 已完成任务计数
    
    // 结果队列：按顺序存储所有任务的 Promise
    let result = [];
    
    // 任务队列：存储正在执行的 Promise
    let queue = [];
    
    let len = tasks.length;
    
    function run() {
        // 执行完所有任务后返回
        if (count === len) {
            return Promise.resolve();
        }
        
        // 获取当前任务
        let task = tasks[count++];
        
        // 每个任务都以 Promise 返回，防止传入的任务非 Promise
        let taskPromise = Promise.resolve(task());
        
        // 将当前正在执行的任务放入【结果队列】
        result.push(taskPromise);
        
        // 当前任务执行完以后，从【任务队列中】删除该任务
        // 这样才能继续往任务队列新加任务
        let _taskPromise = taskPromise.then(() => {
            queue.splice(queue.indexOf(_taskPromise), 1);
        });
        
        // 当前任务放入【任务队列】中
        queue.push(_taskPromise);
        
        let p = Promise.resolve();
        
        // 当任务队列长度达到并发限制数目，暂停往任务队列添加任务
        if (queue.length >= limit) {
            // race 表明只要完成一个任务后，p 就可以执行回调
            // 执行下面的 p.then(() => run())，继续添加任务
            p = Promise.race(queue);
        }
        
        // 注意这里：
        // - 任务队列小于并发限制，跳过 queue.length >= limit，直接执行 p.then 的回调 run 添加任务队列
        // - 任务队列达到并发限制数目后，必须等 Promise.race(queue) 执行完一个任务后，执行 p.then 的回调 run 添加任务队列
        return p.then(() => run());
    }
    
    // 所有任务执行完以后回调，按顺序返回执行结果
    return run().then(() => Promise.all(result));
}

3.2 执行流程详解

graph TD
    A[调用 concurrenceLimits] --> B[初始化变量]
    B --> C[调用 run 函数]
    C --> D{count === len?}
    D -->|是| E[返回 Promise.resolve]
    D -->|否| F[获取任务 tasks count++]
    F --> G[执行任务 task]
    G --> H[将 Promise 加入 result]
    H --> I[将 Promise 加入 queue]
    I --> J{queue.length >= limit?}
    J -->|是| K[Promise.race queue]
    J -->|否| L[Promise.resolve]
    K --> M[等待任意任务完成]
    L --> M
    M --> N[执行 run 继续添加任务]
    N --> D
    E --> O[Promise.all result]
    O --> P[按顺序返回结果]

3.3 关键代码解析

3.3.1 任务执行

let task = tasks[count++];
let taskPromise = Promise.resolve(task());

作用：

获取当前任务并执行
使用 Promise.resolve 确保返回 Promise

3.3.2 结果队列

result.push(taskPromise);

作用：

按顺序存储所有任务的 Promise
最后使用 Promise.all(result) 按顺序返回结果

3.3.3 任务队列管理

let _taskPromise = taskPromise.then(() => {
    queue.splice(queue.indexOf(_taskPromise), 1);
});
queue.push(_taskPromise);

作用：

任务完成后从队列中删除
保持队列中只有正在执行的任务

3.3.4 并发控制

if (queue.length >= limit) {
    p = Promise.race(queue);
}
return p.then(() => run());

作用：

当队列达到并发限制时，等待任意一个任务完成
任务完成后继续执行 run()，添加新任务

四、测试用例

4.1 基础测试

// 模拟请求
function simulateRequest(response, time) {
    const r = `耗时hash_${String(Math.random()).slice(2, 5)}`;
    console.time(r);
    return () => {
        console.log(`开始请求${response}`);
        return new Promise(resolve => setTimeout(() => {
            console.log(`请求${response}: 完成`);
            console.timeEnd(r);
            resolve(response);
        }, time, response));
    };
}

// 测试 DEMO
concurrenceLimits([
    simulateRequest('a', 5000),
    simulateRequest('b', 2000),
    simulateRequest('c', 1000)],
    2).then(res => {
    console.log('result', res);
});

// 控制台输出：
// 开始请求a
// 开始请求b
// 请求b: 完成
// 耗时hash_039: 2.011s
// 开始请求c
// 请求c: 完成
// 耗时hash_114: 3.015s
// 请求a: 完成
// 耗时hash_241: 5.012s
// result [ 'a', 'b', 'c' ]

注意：虽然 b 比 a 先返回，但是最后的结果仍然是按照队列加入顺序来。

4.2 执行时间线

时间轴：
0s    ──> 开始请求 a (5s)
0s    ──> 开始请求 b (2s)
2s    ──> 请求 b 完成，开始请求 c (1s)
3s    ──> 请求 c 完成
5s    ──> 请求 a 完成
      ──> 返回结果 [a, b, c]

五、优化版本

5.1 代码优化

function concurrenceLimits(tasks, limit) {
    if (!Array.isArray(tasks) || tasks.length === 0) {
        return Promise.resolve([]);
    }
    
    if (typeof limit !== 'number' || limit <= 0) {
        throw new TypeError('limit must be a positive number');
    }
    
    const results = [];
    const executing = [];
    let index = 0;
    
    function run() {
        // 所有任务已完成
        if (index >= tasks.length) {
            return Promise.resolve();
        }
        
        // 执行任务
        const task = tasks[index++];
        const promise = Promise.resolve(task())
            .then(result => {
                // 任务完成，从执行队列中移除
                executing.splice(executing.indexOf(promise), 1);
                return result;
            });
        
        results.push(promise);
        executing.push(promise);
        
        // 如果达到并发限制，等待任意一个任务完成
        let racePromise = Promise.resolve();
        if (executing.length >= limit) {
            racePromise = Promise.race(executing);
        }
        
        // 继续执行下一个任务
        return racePromise.then(() => run());
    }
    
    return run().then(() => Promise.all(results));
}

5.2 支持错误处理

function concurrenceLimits(tasks, limit) {
    // ... 前面的代码 ...
    
    function run() {
        if (index >= tasks.length) {
            return Promise.resolve();
        }
        
        const task = tasks[index++];
        const promise = Promise.resolve(task())
            .then(result => {
                executing.splice(executing.indexOf(promise), 1);
                return result;
            })
            .catch(error => {
                executing.splice(executing.indexOf(promise), 1);
                throw error;  // 重新抛出错误
            });
        
        results.push(promise);
        executing.push(promise);
        
        let racePromise = Promise.resolve();
        if (executing.length >= limit) {
            racePromise = Promise.race(executing);
        }
        
        return racePromise.then(() => run());
    }
    
    return run().then(() => Promise.all(results));
}

5.3 支持进度回调

function concurrenceLimits(tasks, limit, onProgress) {
    // ... 前面的代码 ...
    
    function run() {
        if (index >= tasks.length) {
            return Promise.resolve();
        }
        
        const task = tasks[index++];
        const promise = Promise.resolve(task())
            .then(result => {
                executing.splice(executing.indexOf(promise), 1);
                if (onProgress) {
                    onProgress(results.length, tasks.length);
                }
                return result;
            });
        
        // ... 其余代码 ...
    }
    
    return run().then(() => Promise.all(results));
}

六、实际应用场景

6.1 批量请求 API

// 批量获取用户信息
const userIds = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
const tasks = userIds.map(id => () => 
    fetch(`/api/users/${id}`).then(res => res.json())
);

concurrenceLimits(tasks, 3).then(users => {
    console.log('所有用户信息:', users);
});

6.2 文件上传

// 批量上传文件
const files = [file1, file2, file3, file4, file5];
const tasks = files.map(file => () => 
    uploadFile(file)
);

concurrenceLimits(tasks, 2).then(results => {
    console.log('上传结果:', results);
});

6.3 数据爬取

// 爬取多个页面
const urls = ['url1', 'url2', 'url3', 'url4'];
const tasks = urls.map(url => () => 
    fetch(url).then(res => res.text())
);

concurrenceLimits(tasks, 2).then(contents => {
    console.log('爬取结果:', contents);
});