🧊 우선순위 큐로 요청 제어하기

우선순위 큐로 요청 제어하기

분야	작성자	작성일
BE	김민수	24년 12월 03일

개요

현재 가격 상승, 하락률 순위에 대한 API를 요청하여 데이터를 수집하기 위한 기능을 만들었다. 해당 기능에서는 다음의 과정으로 주식 API에 데이터를 요청하고 처리하게 된다.

1. 장이 열리는 시간에 1분 간격으로 상승, 하락률 순위 데이터를 받는다.
2. 총 30개의 종목 중 상위 20개의 데이터를 저장
3. 상위 20개의 실시간 데이터를 요청(1초에 10개 요청)
4. 전달 받은 데이터를 수집 후 저장

이때 주식 API는 계좌 1개 당 1초에 20개의 요청이 가능하기 때문에 적절한 조절이 필요했다. 그로인해 20개의 종목에 대한 데이터를 한번에 받지 않고 1초에 10개씩 나누어서 전달하였고 실패했을 때 2초 대기 후 최대 5번 진행하도록 했다.

async getDecreaseRankStocks(count = 5) {
    try {
      if (count === 0) return;
      const result = await this.getFluctuationRankApiStocks(false);
      const liveResult = await this.getFluctuationRankApiLive(result);
      await this.datasource.transaction(async (manager) => {
        await this.saveFluctuationRankStocks(result, manager);
        await this.saveLiveData(liveResult, manager);
        this.logger.info('decrease rank stocks updated');
      });
    } catch (error) {
      this.logger.warn(error);
      await new Promise((resolve) =>
        setTimeout(() => resolve(this.getDecreaseRankStocks(--count)), 2000),
      );
    }
  }

작성한 코드를 통해 어느 정도 요청 제어가 될 것이라고 생각했다. 하지만 예상외로 생각보다 요청 제어가 잘 이루어지지 않았다.

따라서 하루동안 얼마나 요청이 실패를 확인했다.

놀랍게도 1만 개 이상의 요청이 실패가 되었다. 이렇게 제한된 요청을 잘 제어하지 못한다면 많은 문제가 발생하게 된다. 에러가 계속 발생하여 요청을 보낼 수 있는 기회가 모두 날아가 결국 데이터를 받지 못하는 문제나 불필요한 요청이 늘어나는 문제 그리고 한번 조절을 실패하면 많은 요청이 연쇄적으로 실패할 수 있는 가능성 등 다양한 문제가 발생하게 된다. 따라서 이를 잘 제어할 수 있는 구조를 만들어야된다.

원인

요청 수 제어를 실패한 원인은 현재 요청을 제어하기 위한 로직이 실제로는 제어가 잘 안되기 때문이었다. 기존에 진행한 요청 제어 로직은 크게 2가지였다. 우선 첫 번째는 로직은 다음과 같다.

해당 로직은 스케줄러가 일정 시간이 도달하면 api 요청을 진행하는 로직으로 매우 간단하다. 따라서 1분 주기로 업데이트하는 데이터(변동률 순위, 그와 연관된 라이브 데이터 등)에 쓰이게 된다. 특정 시간에 도달하여 시작하게 된다면 초 당 20개의 요청을 진행하게 된다.

하지만 위 방식의 문제점으로는 각 로직마다 동기화가 제대로 이루어지지 않다는 것이다. 예를 들어서 1분 주기로 변동률 데이터를 업데이트하는 로직을 진행했을 때 동시에 조회수 순 데이터를 업데이트를 하게 된다면, 각 로직에서 요청이 어느 정도 일어나는지 알 수 없기 때문에 결국 조절을 실패하여 에러를 받게 된다.

두 번째 로직은 차트 데이터와 같이 한번에 많은 요청을 제어하기 위해 사용한 로직이다. 해당 방식은 현재 가지고 있는 계좌를 최대한 활용해서 많은 요청을 진행하는 방식이다. 가지고 있는 계좌의 수만큼 Task chunk를 나누어서 setTimeout()을 통해 일정 시간이 지난 후 요청을 진행하도록 했다.

하지만 이것도 요청을 제어하기 힘들었다. 해당 로직의 문제는 첫번째 로직과 같이 다른 요청과 겹치게 될 때 요청이 초과될 수 있는 문제가 있고, setTimeout의 오차가 많이 쌓여 예측하기 어려운 문제가 있었다. 더 큰 문제로 너무 많은 setTimout을 쌓아놓았기 때문에 서버의 메모리가 터지는 문제가 발생하였다. 결국 제한된 요청 수를 잘 제어하기 위해서 새로운 로직을 만들 수 밖에 없었다.

우선순위 큐를 사용하자

해당 문제를 해결하기 위해서 이를 잘 관리할 수 있는 중간 지점이 필요했고, 우선순위 큐를 중간 지점으로 두어 요청 수를 제어할 수 있도록 했다. 개선된 구조는 다음과 같이 생산자 - 큐 - 소비자 형태의 구조를 갖도록 했다.

graph LR
    P1(차트 데이터 스케줄러) --> PriorityQueue
    P2(주가 상세 정보 스케줄러) --> PriorityQueue
    P3(실시간 데이터 요청) --> PriorityQueue
    P4(실시간 데이터 요청) --> PriorityQueue
    PriorityQueue --> C1(계좌1)
    PriorityQueue --> C2(계좌2)
    PriorityQueue --> C3(계좌3)
    PriorityQueue --> C4(계좌4)

Loading

왜 우선순위 큐인가?

큐 대신 우선순위 큐를 선택한 이유는 최대한 실시간 성을 유지하기 위함이다. 각 요청은 사용자에 입장에서 반드시 실시간 성을 유지해야하는 요청과 덜 유지해야하는 요청이 있다.

예를들어서 변동률이나 조회수 순의 데이터 요청의 경우 업데이트가 긴 주기로 발생하기 때문에 실시간 성의 우선순위가 낮다. 하지만, 현재 시점의 주가 정보의 경우 해당 시점의 가격은 실시간 성의 우선순위가 높다. 따라서 우선순위가 높은 요청을 먼저 진행하고, 낮은 요청을 나중에 진행하기 위해 우선순위 큐를 도입했다.

생산자

우선 생산자는 주식 API 데이터 요청을 진행하는 객체로 요청 Task를 큐에 삽입하도록 한다.

• 요청 url - 주가 API에 요청할 url
• 쿼리 - 요청에 필요한 쿼리
• TR_ID - 거래 번호
• callback - 요청이 완료된 데이터를 처리하는 함수

위에 나와있는 항목을 통해 큐의 노드 타입을 정하였다.

interface Json {
  output: Record<string, string> | Record<string, string>[];
  output1: Record<string, string>[];
  output2: Record<string, string>[];
}

interface OpenapiQueueNodeValue {
  url: string;
  query: object;
  trId: TR_ID;
  callback: <T extends Json>(value: T) => Promise<void>;
}

생산자는 요청하고자 하는 API에 맞는 객체를 만들어 큐에 삽입하게 된다.

this.openApiQueue.enqueue({
  url: this.url,
  query,
  trId: TR_IDS.ITEM_CHART_PRICE,
  callback: this.getLiveDataSaveCallback(stock.id!, period),
});

생산자의 요청을 저장하는 자료구조이다. 이때 요청이 실패하거나 우선 순위가 높은 작업을 먼저 빠져나갈 수 있도록 만들었다. 이전에 만든 제너릭 타입의 우선순위 큐를 통해서 Api요청에 대한 로직이 수행할 수 있도록 전용 큐를 만들었다.

class OpenapiQueue {
  private queue: PriorityQueue<OpenapiQueueNodeValue> = new PriorityQueue();

  enqueue(value: OpenapiQueueNodeValue, priority?: number) {
    if (!priority) priority = 2;
    this.queue.enqueue(value, priority);
  }

  dequeue(): OpenapiQueueNodeValue | undefined {
    return this.queue.dequeue();
  }

  isEmpty(): boolean {
    return this.queue.isEmpty();
  }
}

소비자

우선순위 큐로부터 요청을 가져와 처리하는 곳으로 주식 API로부터 데이터를 가져온 후 callback 함수에 처리하도록 한다. 이때 주기적으로 큐가 비었는지 확인하면서 진행하고, 에러가 발생하면 더 높은 우선순위로 재요청을 하도록했다.

class OpenapiConsumer {
  private readonly REQUEST_COUNT_PER_SECOND = 20;
  private isProcessing: boolean = false;
  private currentTokenIndex = 0;

  constructor(
    private readonly queue: OpenapiQueue,
    private readonly openapiTokenApi: OpenapiTokenApi,
    @Inject('winston') private readonly logger: Logger,
  ) {
    this.start();
  }

  async start() {
    setInterval(() => this.consume(), 1000);
  }

  async consume() {
    if (this.isProcessing) {
      return;
    }

    while (!this.queue.isEmpty()) {
      this.isProcessing = true;
      await this.processQueueRequest();
      await new Promise((resolve) => setTimeout(resolve, 1000));
    }
    this.isProcessing = false;
  }

  private async processQueueRequest() {
    const tokenCount = (await this.openapiTokenApi.configs()).length;
    for (let i = 0; i < tokenCount; i++) {
      await this.processIndividualTokenRequest(this.currentTokenIndex);
      if (!this.isProcessing) {
        return;
      }
      this.currentTokenIndex = (this.currentTokenIndex + 1) % tokenCount;
    }
  }

  private async processIndividualTokenRequest(index: number) {
    for (let i = 0; i < this.REQUEST_COUNT_PER_SECOND; i++) {
      const node = this.queue.dequeue();
      if (!node) {
        return;
      }
      this.processRequest(node, index);
    }
  }

  private async processRequest(node: OpenapiQueueNodeValue, index: number) {
    try {
      const data = await getOpenApi(
        node.url,
        (await this.openapiTokenApi.configs())[index],
        node.query,
        node.trId,
      );
      await node.callback(data);
    } catch (error) {
      this.logger.warn(error);
      this.queue.enqueue(node, 1);
    }
  }
}

해당 구조를 적용한 후 이전에 비해서 요청 제어 실패에 대한 에러가 크게 줄어든 것을 볼 수 있다.