Server Sent Event

SSE Flow Sequence Diagram

  SSE는 클라이언트에서 최초에 구독 요청을 보내면 서버가 연결을 끊지 않고 계속 열어둔 채로 데이터를 필요할 때마다 단방향으로 내려보내는 HTTP 표준 프로토콜을 이용하는 실시간 웹 기술입니다.

 

  인메모리에서 서버의 SSE 이벤트를 구독하는 Emitter를 ConcurrentHashMap으로 관리할 경우, 어떤 점에서 메모리 누수가 발생할 수 있고, 동시성을 관리하며 효율을 올릴 수 있을까요?

 

  Java 코드로 해당 부분을 살펴보겠습니다.

콜백 함수

private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

 

  이렇게 Emitter들을 ConcurrentHashMap으로 관리하게 되면 GC는 사용되지 않는 Emitter를 할당 해제하지 못합니다. Map이 emitter 객체들을 강하게 참조(Hard Reference)하고 있기 때문에 더 이상 참조되지 않는 객체가 아니기 때문입니다. 따라서 클라이언트가 떠나도 그대로 객체가 메모리를 잡아먹고 있는 상태로 남겨지게 됩니다. 사용하지 않는 상태에서는 Map의 참조를 반드시 끊어야합니다.

 

 이에 클라이언트가 SSE 구독 요청을 하면 서버에서는 Callback을 통한 즉시 제거를 명시적으로 설정해주어야 합니다.

 

    public SseEmitter connect(Long memberId, LocalDateTime connectTime) {
    	...
        String emitterId = memberId + "_" + timestamp;
        SseEmitter emitter = new SseEmitter(defaultTimeout);

        emitter.onCompletion(() -> {
            log.info("SSE Connection Completed: {}", emitterId);
            emitterRepository.deleteById(emitterId);
        });
        emitter.onTimeout(() -> {
        	emitter.complete();
            log.info("SSE Connection Timed Out: {}", emitterId);
        });
        emitter.onError((e) -> {
            emitter.complete();
            emitterRepository.deleteById(emitterId);
        });

        emitterRepository.save();
        sendBySse();
    }

onCompletion

비동기 요청 처리가 완전히 끝났을 때 무조건 실행되는 콜백 함수입니다.

  1. 개발자가 명시적으로 emitter.complete()를 호출함.
  2. onTimeout 콜백이 실행된 직후에 내부적으로 완료 처리가 되면서 이어서 호출됨.
  3. onError 콜백이 실행된 직후, 내부적으로 완료 처리가 되면서 이어서 호출됨.

onTimeout

  약속된 시간이 지날 때까지 서버가 complete()를 호출하지 않았을 때 실행되는 콜백 함수입니다.

SseEmitter 생성자에서 설정한 시간이 지났을 때 서버에서 리소스를 너무 오랫동안 점유하고 있는 것을 방지하기 위해서 타임아웃을 설정합니다. 이때 타임아웃으로 서버가 연결을 끊으면 브라우저는 자동으로 재연결을 시도하는데, 이 과정에서 Last-Event-ID를 보내서 유실된 데이터를 복구합니다.

 

  즉, onTimeout은 주기적인 리프레시의 기능으로서 활용됩니다.

onError

데이터 전송 중 예외가 발생했거나, 비동기 스레드에서 오류가 났을 때 실행됩니다.

  1. 사용자가 브라우저 탭을 닫음.
  2. 사용자가 새로고침을 함.
  3. 와이파이 등 네트워크가 끊김.

위와 같은 상황 등에서 서버가 emitter.send()를 시도하면 IOException이 발생하면서 onError가 호출됩니다.

혹은 데이터를 생성하는 로직에서 emitter에 영향을 줄 때 onError 콜백이 호출됩니다.

 

SSE 만의 독특한 스펙들

콜백 함수 외에도 SSE 만의 독특한 스펙들은 다음과 같습니다.

Automatic Reconnection

연결이 비정상적으로 끊기면 브라우저는 기본적으로 약 3초 뒤에 자동으로 연결을 재시도합니다.

이때 sse는 retry 필드를 이용해서 서버가 이 재연결 시간을 제어할 수 있습니다.

서버 부하가 심할 때, 이 retry 시간을 늘려서 클라이언트들의 재접속 요청 폭주를 분산시킬 수 있습니다.

HeartBeat

네트워크를 연결했을 때, 이후에 클라이언트가 살아있는지를 지속적으로 파악하기 위해서, 주기적으로 보내는 신호입니다.

연결이 끊긴 Emitter에게 신호를 보내서 죽어있으면 할당을 해제하는 등 메모리를 효율적으로 관리할 수 있습니다. 여러 중계 장치들에서 기본적으로 진행하는 health check도 하트비트에 해당합니다.

 

@Scheduled(fixedRate = 30000)
    public void sendHeartbeat() {
        Map<String, SseEmitter> emitters = emitterRepository.findAll();

        emitters.forEach((id, emitter) -> {
            try {
                emitter.send(SseEmitter.event()
                        .name("heartbeat")
                        .data(""));
                log.trace("Heartbeat sent to: {}", id);
            } catch (IOException | IllegalStateException e) {
                emitter.complete();
            }
        });
    }
}

 

ID 기반의 데이터 복구

재연결은 했는데, 끊긴 동안 데이터가 발생했을 수 있습니다. 다음과 같은 상황으로 예를 들 수 있습니다.

  1. 서버가 주식 시세를 id와 함께 계속 보냄.
  2. 사용자가 엘리베이터를 타서 3초간 네트워크가 끊김.
  3. 서버는 단방향성 이벤트를 송신하므로 사용자가 끊긴 줄 모르기 때문에 계속 시세 데이터를 생성함.

이때 네트워크가 회복되면 브라우저가 자동으로 복구 요청을 합니다.

이는 브라우저 메모리에 lastEventId라는 Internal Slot을 하나 만들어놓고, 데이터 스트림을 파싱하다가 'id' 태그를 발견하면 lastEventId 값을 해당 값으로 갱신하고, 이 값을 retry로 재연결 요청을 보낼때 헤더에 자동으로 추가해서 보냅니다. 이를 서버가 코드 수준에서 처리하면 데이터를 순서에 맞게 복구할 수 있습니다.

@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribe(
    @RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "") String lastEventId
) {
    ...
    if (!lastEventId.isEmpty()) {
        sendLostData(emitter, lastEventId);
    }
    ...
}

 

동시성

    public Map<String, SseEmitter> findAllStartWithById(Long memberId) {
        Map<String, SseEmitter> result = new ConcurrentHashMap<>();
        emitters.forEach((id, emitter) -> {
            if(id.startsWith(memberId.toString())){
                result.put(id, emitter);
            }
        });
        return result;
    }

memberId로 해당하는 emitter를 찾는 메서드

    public void sendToMembers(Map<String, Object> msg, List<Long> receiverIds){
        for (Long receiverId : receiverIds) {
            Map<String, SseEmitter> emitters = emitterRepository.findAllStartWithById(receiverId);

            emitters.forEach((id, emitter) -> {
                sendBySse(emitter, id, "notification", msg);
            });
        }
    }

receiverIds에게 sse event를 send하는 메서드

    private void sendBySse(SseEmitter emitter, String id, String eventName, Object data) {
            emitter.send(SseEmitter.event()
                    .id(id)
                    .name(eventName)
                    .data(data));
    }

send 메서드

 

  1. Thread A: 알림 발송을 시작함
  2. Thread B: Emitter1의 연결이 종료됨
  3. Thread A: 알림 발송을 위해서 findAllStartWithById()를 요청함 (Emitter 1이 포함된 Map을 순회)
  4. Thread: B: Emitter1 삭제 완료됨
  5. Thread A: Emitter 1에게 알림을 발송함 -> 이미 끊어진 연결에 전송 시도 -> IO Exception

서버 측 상황

  만약에 try-catch로 해당 경우를 격리하지 않으면 메서드 자체에서 Exception이 발생합니다. 여기서 Java에서는 emitters.forEach 안에 있는 emitter.send() 메서드를 강제로 종료시킵니다. 그럼 forEach 문도 그 즉시 멈추게되고, 리스트에 그 뒷 순서로 남아있던 나머지 클라이언트는 알림을 받지 못하게됩니다. 즉, 일부에게만 알림이 보내집니다. 이제 try catch를 적용해보겠습니다.

    private void sendBySse(SseEmitter emitter, String id, String eventName, Object data) {
        try {
            emitter.send(SseEmitter.event()
                    .id(id)
                    .name(eventName)
                    .data(data));
        } catch (IOException | IllegalStateException e) {
            throw e;
        }
    }

 

  이제는 IO Exception: Broken pipe의 에러가 발생하면 그 스레드의 작업은 즉시 중단되고 catch 블록으로 점프합니다. 여전히 에러를 던지므로 에러를 잡아냈지만 위와 같은 상황이 계속됩니다.

 

  위 둘의 상황에서 emitters 객체에서는 이때 연결은 끊겼지만 해당 emitter는 힙 메모리의 emitters 객체 안에 연결돼있어서 GC의 대상에서 제외되어 메모리를 계속해서 차지합니다. 또한 다음번에 알림을 보낼 때 또 에러가 발생합니다.

    private void sendBySse(SseEmitter emitter, String id, String eventName, Object data) {
        try {
            emitter.send(SseEmitter.event()
                    .id(id)
                    .name(eventName)
                    .data(data));
        } catch (IOException | IllegalStateException e) {
            emitterRepository.deleteById(id);
        }
    }

 

따라서 에러를 던지지 않고, 해당하는 emitter를 삭제하면 메모리 누수를 막고 이어지는 에러를 막을 수 있습니다.

클라이언트 측 상황

연결이 끊겼으므로 retry를 진행합니다. 여기서 Last-Event-ID 헤더를 자동으로 붙여서 복구를 진행할 수 있습니다.

 

데이터 정합성

  Last-Event-ID는 서버에서 보내는 'id' 태그의 내용을 브라우저에서 저장해서 가장 최근의 내용을 헤더에 injection해서 retry를 진행하는 방식입니다. 여기서 서버에서는 적절한 id 태그의 내용과, 이 기록들을 얼마나 저장해놓을 것인가를 설계해야 합니다.

 

  지금 개선하고 있는 서비스는 SSE로 사용자가 연결이 되어 있을 때 알림이 도착하면 '새로운 알림이 도착했습니다!'를 토스트 메시지로 띄워주는 간단한 기능을 가지고 있습니다. 여기서 데이터가 유실되는 것을 막으면서도 사용자가 알림에 관심을 가질 수 있도록 'N개의 알림이 도착했습니다!'를 띄워줍니다.

 

id 태그에 memberId + id을 넣어서 사용자에게 전송하다가 중간에 커넥션이 끊겨서 데이터가 유실되면 id을 이용해서 만약 끊긴 클라이언트가 접속할 경우, 'N개의 새 알림이 도착했습니다!' 라고 메시지를 수정해서 제시합니다.

 

*실제 코드에서는 다양한 추상화 속에서 위 코드를 캡슐화해서 구성했습니다. 또한 메시지에 대한 util 함수와 enum을 통한 관리에 병합해서 구현했습니다.

 

❓ onError나 onTimeout을 호출하면 내부적으로 onCompletion을 호출한다면서? 근데 코드에서는 코드에서 호출하는거지?

더보기

  위 코드에서는 onError와 onTimeout을 호출할 때도 명시적으로 onCompletion을 호출하는데요, 이는 onTimeout 로직이 끝났는데 응답에 대한 결론 (완료, 에러 등) 의 처리가 없으면 에러로 처리될 수 있습니다.

 

  서블릿 컨테이너에서는 시간은 다 됐는데, 애플리케이션이 종류 처리를 안 했다고 판단해서 이를 503 에러로 처리하고 연결을 강제로 끊습니다.

 

  여기서 직접 complete()을 명시적으로 호출하면 에러 처리가 되지 않으며 호출 그 즉시 리소스 정리 단계로 넘어갑니다. 호출이 없는 경우에는 서블릿 컨테이너가 개입하는 등의 오버헤드가 발생하면서 정리되면서 효율을 떨어뜨립니다.

반응형

'Backend > Spring' 카테고리의 다른 글

[Spring] @Transactional의 구조와 Retry 정책 구현  (0) 2026.02.19
[Spring] Spring에서의 Tomcat과 비교한 Netty의 구조  (0) 2026.02.11

티스토리툴바