🔌 websocket이 늦게 할당되어 발생되는 문제

분야	작성자	작성일
BE	김성환	24년 12월 04일

들어가기 전

• 주춤주춤 서비스는 한국투자증권의 웹소켓을 사용해 주식의 현재 체결량과 같은 실시간 데이터를 받을 수 있다.

• 유저에게 현재가를 보여주기 위해서는 주식마다 라이브데이터를 구독하는 것이 최고지만, 웹소켓을 만들 수 있는 개수와 주식을 할당하는 개수도 한정되어 있다

  • 현재 웹소켓을 한국투자증권과 연결할 수 있는 개수는 4개, 주식을 할당할 수 있는 개수는 웹소켓당 41개이다.
  • 따라서 164개의 데이터를 라이브로 제공할 수 있으나, 우리나라의 주식은 대략 4200개정도로 턱없이 부족하다.

• 문제를 해결하기 위해 유저와 연결은 socket.IO의 room기능을 활용해 현재 주식창을 보고 있는 지 체크하고, 한국투자증권과의 연결은 Websocket을 이용해 연결하는 방식으로 구현했다.

  • 유저가 주식창을 보고 있으면 한국투자증권에서 주식 정보를 받을 수 있게 구독하고, 한 명의 유저도 보지 않는 다면 주식 정보 구독을 취소하는 방식으로 제한되어 있는 구독 정보를 효율적으로 관리했다.

문제 상황

• 유저가 접속했을 때, 서버가 막 켜졌을 상황일 경우 라이브데이터가 제대로 받아지지 못하는 경우가 발생했다.
  • 실시간 데이터가 받아지지 않는 문제가 발생했다.

각 클래스 설명

• LiveData

  • 한국투자증권 웹소켓의 메시지, 리턴 메시지 등 비즈니스 로직을 처리하는 클래스
  • 4개의 콜백함수를 WebsocketClient 클래스에 할당하는 방식으로 작동한다.
  • WebsocketClient를 할당할 수 있는 개수가 런타임 때 결정되어 이를 WebsocketClient 를 동적으로 할당한다.

• WebsocketClient

  • 웹소켓 로직만 처리하는 클래스
  • 로깅, 연결을 하고, 콜백을 외부에서 주입받는 클래스

이전 코드

export class WebsocketClient {
  static url = process.env.WS_URL ?? 'ws://ops.koreainvestment.com:21000';
  private client: WebSocket;

  constructor(@Inject('winston') private readonly logger: Logger) {
    this.client = new WebSocket(WebsocketClient.url);
  }
  
  // 중간 생략
  private sendMessage(message: string) {
    if (this.client.readyState === WebSocket.OPEN) {
      this.client.send(message);
      this.logger.info(`Sent message: ${message}`);
    } else {
      this.logger.warn('WebSocket not open. Queueing message.');
    }
  }
  
  static websocketFactory(logger: Logger) {
    const websocket = new WebsocketClient(logger);
    return websocket;
  }
  //이하 생략
}

• 이런 환경에서 유저가 subscribe를 해서 주식 데이터를 받고자 하면 다음 sequence diagram 처럼 문제가 발생한다.

sequenceDiagram
    participant User
    participant LiveData
    participant WebsocketClient
    participant WebsocketToKoreaInvestment

		LiveData->>WebsocketClient: 동적으로 할당하는 중
    User->>LiveData: subscribe()
    LiveData->>WebsocketClient: subscribe()
    WebsocketClient->>WebsocketToKoreaInvestment: (아직 소켓이 열리지 않은 상태)
    WebsocketClient->>User: Error: socket not assigned

Loading

원인

• Websocket내 코드에선 대부분 동기로 작동하지만, upgrade요청에 대해서 event emitter를 통해 받아서 처리한다.

// Websocket lib/websocket.js
req.on('upgrade', (res, socket, head) => {
    websocket.emit('upgrade', res);
    //중간 생략
    
    // socket을 실질적으로 할당하는 함수
    // 할당이 완료되면 _readyState를 WebSocket.OPEN으로 변경
    websocket.setSocket(socket, head, {
      allowSynchronousEvents: opts.allowSynchronousEvents,
      generateMask: opts.generateMask,
      maxPayload: opts.maxPayload,
      skipUTF8Validation: opts.skipUTF8Validation
    });
    //이하 생략
 })

• 웹소켓의 경우 처음에 http 또는 https 로 먼저 연결을 시도하고, 101 switching protocols이 들어오면 http에서 Websocket 으로 업그레이드를 한다.

• 이때, event emitter를 이용해 upgrade 요청을 처리하고, 한국투자증권와의 소켓이 열린 상황이 아닌 상태에서 유저의 요청이 들어올 수 있다는 점이었다.

해결책 1 - OnModuleInit

• nestjs의 lifecycle에 맞춰 문제를 해결하는 방식이다.
• 다만, 이런 구조에도 실질적으로 소켓이 언제 할당되는 지는 알 수 없으므로 문제를 완벽하게 해결할 순 없었다.

해결책 2 - 큐를 이용해 문제를 해결

• 문제를 실질적으로 해결하기 위해선 유저의 요청을 임시로 저장하고, 만약 소켓이 열린다면 그 이후 요청을 처리하는 로직이 필요했다.
• 이때 큐를 어디에 적용하는 지에 대한 고민이 있었다.
  • LiveData - 비즈니스 로직을 처리하는 부분으로 웹소켓의 연결과는 상관 없는 로직만 처리하는 것을 처음에 생각했었다.
  • 따라서 큐를 적용해 유저 요청을 처리하는 것에는 WebsocketClient에 적용하는 것이 맞다고 생각했다.

export class WebsocketClient {
  static url = process.env.WS_URL ?? 'ws://ops.koreainvestment.com:21000';
  private client: WebSocket;
  private messageQueue: string[] = [];

  constructor(@Inject('winston') private readonly logger: Logger) {
    this.client = new WebSocket(WebsocketClient.url);
    this.initOpen(() => this.flushQueue());
    this.initError((error) => this.logger.error('WebSocket error', error));
  }

  static websocketFactory(logger: Logger) {
    return new WebsocketClient(logger);
  }
  
  //중간 생략
  
  private sendMessage(message: string) {
    if (this.client.readyState === WebSocket.OPEN) {
      this.client.send(message);
      this.logger.info(`Sent message: ${message}`);
    } else {
      this.logger.warn('WebSocket not open. Queueing message.');
      this.messageQueue.push(message); // 큐에 메시지를 추가
    }
  }

  private flushQueue() {
    while (this.messageQueue.length > 0) {
      const message = this.messageQueue.shift();
      if (message) {
        this.sendMessage(message);
      }
    }
  }
}

• 이런 방식으로 실제로 소켓이 열리지 않은 상황에서도 안정적으로 유저의 실시간 데이터 요청에 응답할 수 있었다.

참고자료

Websocket github

nestjs lifecycle

websocket connection