주식관련 메신저 프로젝트를 진행하면서, 실시간 알림기능을 맡게 되었습니다.

현제 구상한 알림 기능으로는 내가 즐겨찾기한 주식이 변동할때 알림기능과

톡방에서 상대방이 메세지를 보냈을때 실시간으로 알림이 오는 기능입니다.

그래서 저는 실시간 알림 기능에 대해서 찾아봤습니다.

우선, HTTP 프로토콜에서는 client에서 요청이 있어야 server에서 응답을 보낼 수 있습니다.
client에 요청없이 server에서는 응답을 보내는 것은 불가능합니다.

실시간 통신에 대해 찾아보면서

웹 소켓에 대해서 알게 되었습니다.

웹 소켓(Web Socket)

웹 소켓이란 두 프로그램 간의 메시지 교환을 위한 통신 방법입니다.

웹 소켓의 특징

• 양방향 통신(Full-Duplex)

• 실시간 네트워킹

• http 프로토콜이 아닌 ws 프로토콜 사용

웹 소켓의 특징으로 실시간 채팅알림, 또는 실시간으로 변화하는 주식장에서 사용자가 선택한 매수/매도 타이밍에 알림에 대해 적합하다고 생각하였습니다.

하지만, 이러한 장점이 있는 웹 소켓을 선택하지 않은 이유로는
저는 실시간 채팅을 맡은게 아닌 알림 기능만 맡았기 때문에 굳이 2개의 웹 소켓을 열 필요가 없다고 생각했습니다.

또한 웹 소켓은 한번 연결되면 별도의 에러나, 지시가 없는한 지속적으로 연결을 유지하므로 서버의 부하가 증가할 수 있기 때문입니다.

SSE(Server Sent Events)

SSE는 단방향 통신이며 클리이언트의 별도 추가요청 없이 서버에서 업데이트를 스트리밍할 수 있다는 특징을 가집니다.

SSE 특징

• 서버에서 실시간으로 이벤트 전송

• polling 기법보다 적은 통신 횟수

• 새로운 프로토콜을 익힐 필요가 없음

• 서버 측에서의 단방향 통신

SSE 통신 과정

1. SSE 구독 신청

클라이언트가 서버와의 연결을 요청하는 것을 구독이라는 행위로 나타낼 수 있다. 서버의 변화를 구독하여 실시간으로 관찰하고 싶다는 의미이다.

    @GetMapping(path = "/emitter", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
   public SseEmitter subscribe() {
     
    }

컨트롤러에는 produces로 Content-Type을 MediaType.TEXT_EVENTSTREAM_VALUE 으로 지정하였다.
이렇게 지정하면 서버에서 응답 패킷의 헤더의 Content-Type이 text/event-stream으로 되고 텍스트 데이터를 연속해서 보낼 수 있다.

2. SSE 연결 수립

• SSeEmitter 객체 생성
  가장 먼저 클라이언트의 요청에 따라 서버에서는 통신 객체인 SSeEitter를 생성한다. 생성할때 만료 시간을 입력할 수 있다. 만료시간이 너무 길면 연결 관리를 해줘야하고 짧으면 재연결 요청이 잦게 일어나므로 적절하게 설정 해야한다.

            SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

• 더미 데이터 전송
  서버와의 연결을 맺을때 더미 데이터를 넘겨줘야 한다. 그렇지 않고 데이터를 하나도 전송하지 않으면 재연결 요청시 503 Service Unavaliable 에러가 발생한다. 따라서 이를 방지하기 위해 Dummy 데이터를 보낸다.
  

   //주기적으로 이벤트를 보내는 예시 (필요하지 않다면 삭제 가능)
    @Scheduled(fixedRate = 1000)
    public void sendEvents() {
       for (SseEmitter emitter : emitters) {
            try {
                emitter.send("Hello, World!");
           } catch (IOException e) {
               emitter.complete();
                emitters.remove(emitter);
           }
       }
    }

• Emitter 객체 관리
  Emitter를 저장하기 위해 Spring Data JPA를 사용해서 물리적으로 저장해도 된다. 하지만 HashMap을 사용한 케이스가 많아 HashMap을 사용했다. HashMap을 사용하면 데이터 저장 위치를 해시함수를 통해 바로 알 수 있어 검색이 빠르다. 일반 HashMap은 Thread-safe 하지 않으므로 ConcurrentHashMap을 사용한다. 멀티 쓰레드 환경에서 데이터를 관리하기 위해 동기화 과정이 필요하다. 이를 위해 쓰레드가 접근하는 Map에 Lock을 걸어야하는데 ConcurrentHashMap 에서는 Bucket 단위로 관리가 되어 오버헤드를 줄일 수 있다는 장점이 있다.
  

    private final List<SseEmitter> emitters = new CopyOnWriteArrayList<>();

3. 클라이언트에게 메세지 보내기

마지막으로 연결되어있는 클라이언트에게 메세지를 전달하는 컨트롤러를 만들었다.
sendData에 전달할 message를 인자로 주어 메세지를 전달한다.

    // POST 요청을 통해 데이터를 수신하고, 이를 SSE로 클라이언트에 전달
    @PostMapping("/send")
    public void sendData(@RequestBody String message) {
        // 받은 데이터를 모든 연결된 클라이언트에 전송
        sseService.sendMessageToClients(message);
    }