[node.js] 동시에 실행되는 수를 조절하며 비동기 함수 전체 실행 (Promise.all with limited concurrency)

 

Promise.all()에 동시에 실행되는 promise의 수를 조절하는게 없어서 만들었음.

예를 들어, 사용자 10000명에게 동시에 메시지를 전송해야 할 경우,
하나씩 하나씩 보낼 수는 없으니 비동기 함수를 동시 실행해야 할텐데,
너무 많은 수의 비동기 함수를 Promise.all()로 한 번에 때리면 문제가 일어날 가능성이 있으므로,
동시에 10명씩 혹은 100명씩 이렇게 잘라서 보내고 싶을 때 필요한 것.
비슷한 일을 할 수 있는 모듈들(async, p-limit, bluebird 등)이 있으나 모듈 추가되는게 싫어서 만들었음.

module.exports = {
    // itemList: 처리해야 할 item들이 담긴 Array
    // limit: 동시 실행 수
    // fnPromise: itemList의 item 하나를 받아 처리할 비동기 함수
    item: (itemList, limit, fnPromise) => new Promise(async(resolved, rejected)=>{
        if (typeof itemList !== 'object' || !('length' in itemList)) {
            return rejected('itemList must be Array');
        }
 
        limit = parseInt(limit);
        if (limit < 1) {
            return rejected('limit must be greater than zero');
        }
 
        if (typeof fnPromise !== 'function') {
            return rejected('fnPromise must be a function returns Promise');
        }
 
        const resultList = [];
 
        const itemListLen = itemList.length;
        let taskList = [];
        for (let i=0; i<limit; i++) {
            if (i >= itemListLen) break;
 
            taskList.push(new Promise(async(resolved, rejected)=>{
                while (true) {
                    const item = itemList.shift();
                    if (!item) {
                        return resolved(null);
                    }
 
                    try {
                        resultList.push(await fnPromise(item));
                    } catch (e) {
                        return rejected(e);
                    }
                }
            }));
        }
 
        try {
            await Promise.all(taskList);
        } catch (e) {
            return rejected(e);
        }
 
        return resolved(resultList);
    }),
 
    // asyncFnList: 실행될 비동기 함수들이 담겨 있는 Array
    // limit: 동시 실행 수
    func: (asyncFnList, limit) => new Promise(async(resolved, rejected)=>{
        if (typeof asyncFnList !== 'object' || !('length' in asyncFnList)) {
            return rejected('asyncFnList must be Array');
        }
 
        limit = parseInt(limit);
        if (limit < 1) {
            return rejected('limit must be greater than zero');
        }
 
        const resultList = [];
 
        let taskList = [];
        for (let i=0; i<limit; i++) {
            taskList.push(new Promise(async(resolved, rejected)=>{
                while (true) {
                    const task = asyncFnList.shift();
                    if (!task) {
                        return resolved(null);
                    }
 
                    try {
                        resultList.push(await task());
                    } catch (e) {
                        return rejected(e);
                    }
                }
            }));
        }
 
        try {
            await Promise.all(taskList);
        } catch (e) {
            return rejected(e);
        }
 
        return resolved(resultList);
    })
};

 

 

위의 소스코드가 async_limit.js 라는 파일에 저장되어 있다고 가정하고 아래처럼 사용하면 됨.

const AsyncLimit = require('./async_limit.js');
 
 
 
 
(async()=>{
 
// 이건 그냥 시간 재보기 위해 만든 Object
const StopWatch = {
    startTime: null,
 
    start: function(){
        this.startTime = process.hrtime.bigint();
    },
 
    lap: function(){
        return parseInt(process.hrtime.bigint() - this.startTime) / 1000000;
    }
};
 
 
// 100ms 쉬고 return resolved() 하는 비동기 함수 10개를 만들어 Array에 저장
const asyncFnList = [];
for (let i=0; i<10; i++) asyncFnList.push(()=>{
    const started = StopWatch.lap();
 
    return new Promise((resolved)=>{
        setTimeout(()=>{
            const ended = StopWatch.lap();
            const took = ended - started;
            return resolved(`started: ${started.toFixed(3)}ms, ended: ${ended.toFixed(3)}ms, took ${took.toFixed(3)}ms`);
        }, 100);
    });
});
 
StopWatch.start();
 
// 비동기 함수 배열을 동시에 3개씩 실행되도록 실행
let result = await AsyncLimit.func(asyncFnList, 3);
 
console.log('[AsyncLimit.item]');
result.forEach((r)=>{
    console.log(r);
});
 
/*
결과
[AsyncLimit.func]
started: 0.940ms, ended: 103.426ms, took 102.485ms
started: 2.556ms, ended: 103.864ms, took 101.308ms
started: 2.602ms, ended: 103.880ms, took 101.278ms
started: 104.151ms, ended: 204.560ms, took 100.409ms
started: 104.269ms, ended: 204.580ms, took 100.311ms
started: 104.277ms, ended: 204.584ms, took 100.307ms
started: 204.603ms, ended: 304.829ms, took 100.226ms
started: 204.624ms, ended: 304.851ms, took 100.227ms
started: 204.629ms, ended: 304.858ms, took 100.230ms
started: 304.967ms, ended: 405.428ms, took 100.461ms
 
 
 
 
limit를 4로 했을 경우
[AsyncLimit.func]
started: 0.914ms, ended: 102.868ms, took 101.955ms
started: 2.742ms, ended: 104.663ms, took 101.921ms
started: 2.787ms, ended: 104.677ms, took 101.890ms
started: 2.832ms, ended: 104.682ms, took 101.850ms
started: 103.502ms, ended: 202.943ms, took 99.440ms
started: 104.690ms, ended: 205.139ms, took 100.449ms
started: 104.697ms, ended: 205.149ms, took 100.452ms
started: 104.701ms, ended: 205.153ms, took 100.452ms
started: 202.991ms, ended: 302.710ms, took 99.718ms
started: 205.266ms, ended: 305.878ms, took 100.611ms
*/
 
 
// 100을 Array에 10개 저장
const timeList = [];
for (let i=0; i<10; i++) timeList.push(100);
 
StopWatch.start();
 
// 100이 10개 저장된 Array와 Array의 item을 각각 처리하는 핸들러를 넘기는 방식
result = await AsyncLimit.item(timeList, 3, (time)=>{
    const started = StopWatch.lap();
 
    return new Promise((resolved)=>{
        setTimeout(()=>{
            const ended = StopWatch.lap();
            const took = ended - started;
            return resolved(`started: ${started.toFixed(3)}ms, ended: ${ended.toFixed(3)}ms, took ${took.toFixed(3)}ms`);
        }, time);
    });
});
 
console.log('[AsyncLimit.func]');
result.forEach((r)=>{
    console.log(r);
});
 
 
/*
결과
[AsyncLimit.item]
started: 0.391ms, ended: 100.566ms, took 100.175ms
started: 0.456ms, ended: 100.597ms, took 100.141ms
started: 0.505ms, ended: 100.605ms, took 100.100ms
started: 100.632ms, ended: 200.830ms, took 100.197ms
started: 100.654ms, ended: 200.850ms, took 100.196ms
started: 100.658ms, ended: 200.855ms, took 100.197ms
started: 200.870ms, ended: 301.082ms, took 100.212ms
started: 200.891ms, ended: 301.104ms, took 100.213ms
started: 200.895ms, ended: 301.212ms, took 100.317ms
started: 301.233ms, ended: 401.462ms, took 100.229ms
 
 
 
 
limit를 2로 했을 경우
[AsyncLimit.item]
started: 0.297ms, ended: 101.081ms, took 100.784ms
started: 0.454ms, ended: 101.111ms, took 100.657ms
started: 104.806ms, ended: 204.932ms, took 100.125ms
started: 104.831ms, ended: 204.951ms, took 100.120ms
started: 204.967ms, ended: 305.157ms, took 100.189ms
started: 204.987ms, ended: 305.178ms, took 100.191ms
started: 305.195ms, ended: 405.404ms, took 100.209ms
started: 305.217ms, ended: 405.425ms, took 100.208ms
started: 405.440ms, ended: 505.638ms, took 100.197ms
started: 405.464ms, ended: 505.658ms, took 100.195ms
*/
 
})();