并发请求调度池

2019-07-18

问题

请实现如下的函数，可以批量请求数据，所有的url地址在urls参数中，同时可以通过max参数控制请求的并发度，当所有请求结束之后，需要执行callback回调函数。发请求的函数可以直接使用fetch即可。

1
2
3

function sendRequest(urls: string[], max: number, callback: () => void) {
    // TODO
}

问题分析

本题目关键在于抓住关键点

批量请求：要实现批量请求，而且并不需要按顺序发起请求（如果需要按顺序可以存入队列中，按优先级则可以存入优先队列中），这块儿存入数组中即可，然后遍历，取出每项fetch调用。
可控制并发度：在控制并发数的同时，每结束一个请求并发起一个新的请求。这里维护一个请求队列，使用递归的方式每次发起请求就推入队列，请求结束时从队列抛出。
全部请求结束，执行callback：每fetch回调then或者catch时，判断请求数组是否为空就来判断是否全部被调用完

代码实现

function sendRequest(urls: string[], max: number, callback: () => void) {
    const urlCount = urls.length;
    const requestsQueue = [];
    const results = [];
    let i = 0;
    const handleRequest = url => {
        const req = fetch(url).then(res => {
            const len = results.push(res);
            // 请求结果数组的长度小于整个需要请求的大数组的长度（还有要进行的任务） && 下一条请求存在
            if (len < urlCount && i + 1 < urlCount) {
                // 当前任务执行完毕，从请求队列丢出
                requestsQueue.shift();
                // 1、i自加 2、执行下一条请求
                handleRequest(urls[++i]); // ?
            // 请求全部调用完成，执行callback
            } else if (len === urlCount) {
                callback && callback(results);
            }
        }).catch(e => {
            results.push(e);
        });
        if (requestsQueue.push(req) < max) {
            handleRequest(urls[++i]);
        }
    };
    handleRequest(urls[i]);
}

// 为了数据方便，这里模拟一个fetch请求
const fetch = function (idx) {
  return new Promise(resolve => {
    console.log(`start request ${idx}`);
    const timeout = parseInt(Math.random() * 1e4);
    setTimeout(() => {
      console.log(`end request ${idx}`);
      resolve(idx);
    }, timeout)
  });
};

调用：

const urls = Array.from({ length: 10 }, (v, k) => k);
const max = 4;
sendRequest(urls, max, function() {
    console.log('callback runed');
});

效果如下：

代码改进 async/await

function sendRequest(urls: string[], max: number, callback: () => void) {
    const urlCount = urls.length;
    const requestsQueue = [];
    const results = [];
    let i = 0;
    const handleRequest = url => {
        const req = fetch(url).then(res => {
            const len = results.push(res);
            // 请求结果数组的长度小于整个需要请求的大数组的长度（还有要进行的任务） && 下一条请求存在
            if (len < urlCount && i + 1 < urlCount) {
                // 当前任务执行完毕，从请求队列丢出
                requestsQueue.shift();
                // 1、i自加 2、执行下一条请求
                handleRequest(urls[++i]); // ?
            // 请求全部调用完成，执行callback
            } else if (len === urlCount) {
                callback && callback(results);
            }
        }).catch(e => {
            results.push(e);
        });
        if (requestsQueue.push(req) < max) {
            handleRequest(urls[++i]);
        }
    };
    handleRequest(urls[i]);
}

function async query(url: string, data?) {
    const body = qs.stringify(data);
    let res;
    try {
        res = await axios.post(url, body);
    } catch (e) {
        console.error(e);
    }
    return res && res.data;
}

// 为了数据方便，这里模拟一个fetch请求
const fetch = function (idx) {
  return new Promise(resolve => {
    console.log(`start request ${idx}`);
    const timeout = parseInt(Math.random() * 1e4);
    setTimeout(() => {
      console.log(`end request ${idx}`);
      resolve(idx);
    }, timeout)
  });
};

调用：

const urls = Array.from({ length: 10 }, (v, k) => k);
const max = 4;
sendRequest(urls, max, function() {
    console.log('callback runed');
});