기본으로 돌아가자_동시실행제한큐

안승찬·2026년 6월 1일

이번에는 시스템 모듈에 많이 사용되는 큐 방식에 대해서 구현을 해볼려고 합니다.

────
결제 시스템, 대량 알림 발송, 이미지 업로드 등에서
"동시에 N개만 실행하고 나머지는 순서대로 대기"가 필요합니다.
Promise.all은 모두 동시에 터뜨리기 때문에 서버 부하·Rate Limit에
걸릴 수 있습니다.

목표
────
ConcurrentQueue 클래스를 완성하세요.

스펙
────
constructor({ concurrency })
	- concurrency: 동시에 실행할 최대 작업 수 (기본값 1)

add(task) → Promise
  - task: () => Promise 형태의 함수
  - 실행 슬롯이 남아 있으면 즉시 실행, 아니면 대기열에 추가
  - task가 resolve/reject 되면 그 값을 그대로 전파
  - 한 task가 reject 되어도 다른 task 실행에 영향을 주지 않음

pause()
  - 대기 중인 작업의 시작을 멈춤
  - 이미 실행 중인 작업은 끝까지 실행됨

resume()
  - pause 해제, 대기 중인 작업을 concurrency에 맞게 재개

clear()
  - 대기 중인 작업을 모두 제거
  - 이미 실행 중인 작업은 영향 없음
  - 제거된 작업의 Promise는 reject({ reason: 'cleared' }) 됨

 get metrics()
  - { running, queued, completed, failed } 반환

먼저 구조를 잡아 봅시다.
생성자로는 작업을 담을 queue와 동시에 작업할 수 있는 concurrency가 먼저 생각났습니다.

class ConcurrentQueue {
    constructor({concurrency = 1}={}){
      this.queue:[];
      this.concurrency=concurrency;
    }
  
  	add(task){
      
    }
  
  	pause() {

	}

	resume() {

    }

    clear() {
      
    }

	get metrics() {
    
	}
  
  
}

그 다음 요구사항 위에서 부터 구현해 보도록 하겠습니다.

...
constructor({concurrency = 1}={}){
      this.queue = [];
      this.concurrency=concurrency;
	  this.runningCount=0;//실제 실행되고 있는 작업 수
    }


add(task){
	if (typeof task !== "function")  throw new TypeError("task must be a function");
  //task: () => Promise 형태
  return new Promise((resolve,reject)=>{
    //task와 실행결과를 큐에 담는다
    this.queue.push({task,resolve,reject});
     nextTask();
  })
}
  
nextTask(){
    // 실행 가능한 작업 수를 넘겼거나 실제 작업 큐가 비었을 경우
    if(this.runningCount >= this.concurrency || this.queue.length===0) return;
    
     // 맨 첫번째 대기열 작업 
	 const next = this.queue.shift();
     this.runningCount++;
     
     next.task().then((result)=>{
       //작업 성공
       next.resolve(result)
     }).catch((error)=>{
       //작업 실패
       next.reject(error)
     }).finally(()=>{
       //작업 테스크 카운트 감소
       this.runningCount--;
       //다음 작업 진행
       this.nextTask();
     })
}

순환 구조를 잡는 부분 제외하고는 나머지부분은 어렵지 않게 구현 했던 것 같습니다.

그 다음으로 나머지 함수들도 만들어 봅시다.

 constructor({concurrency = 1}={}){
      this.queue = [];
      this.concurrency=concurrency;
	  this.runningCount=0;
  	  this.isPaused=false; // 일시정지 상태 여부
    }

 	...

      nextTask(){
      // 중지 상태 이면 다음 작업 진행 X
      if(this.isPaused) return;
      if(this.runningCount >= this.concurrency || this.queue.length===0) return;

       const next = this.queue.shift();
       this.runningCount++;

       next.task().then((result)=>{
         next.resolve(result)
       }).catch((error)=>{
         next.reject(error)
       }).finally(()=>{
         this.runningCount--;
         this.nextTask();
       })
  }

	pause() {
	   this.isPaused =true;
	}

	resume() {
	   this.isPaused =false;
	   this.nextTask();
    }

    clear() {
       while(this.queue.length>0){
         const item = this.queue.shift();
         // 모두 실패 처리
         item.rejected({reason : "cleared"})
       }
      
       // 큐 초기화
       this.queue=[];      
    }

검증

 const queue = new ConcurrentQueue({
    concurrency: 2
  });
  
  const task = (name, delay) => () =>
    new Promise((resolve) => {
      console.log("start", name);
  
      setTimeout(() => {
        console.log("end", name);
        resolve(name);
      }, delay);
    });
  
  queue.add(task("A", 1000));
  queue.add(task("B", 1000));
  queue.add(task("C", 1000));
  queue.add(task("D", 1000));


---console---
start A
start B
end A
start C
end B
start D
end C
end D