Test Results

33 tests   33 ✅  5s ⏱️
14 suites   0 💤
14 files     0 ❌

Results for commit 40bc606.

♻️ This comment has been updated with latest results.

지금까지 겪은 문제

SseEmitters 동시성 문제

SseEmitterStorage는 클라이언트와 연결된 SSE 커넥션들을 관리하는 인메모리 자료구조입니다. 내부적으로 어떤 자료구조를 사용하는지 결정해야 합니다.

// 현재 pr 버전
@Component
public class SseEmitterStorage {

    /*
        ExecutorService로 변경이 필요할 수 있습니다.
     */
    private final ConcurrentLinkedQueue<SseEmitter> emitters;

    /*
        동시성 문제로 자료구조 변경이 필요할 수 있습니다.
     */
    public SseEmitterStorage() {
        this.emitters = new ConcurrentLinkedQueue<>();
    }

처음에는 ArrayList를 사용했어요. 그랬더니 동시성 예외가 발생했어요.

이유는 다음 작업이 동시에 일어나기 때문이에요.

• emitters에서 종료된 SseEmitter 객체를 삭제하는 로직

// sseStorage
public void add(SseEmitter sseEmitter) {
    emitters.add(sseEmitter);
    log.info("[SSE] 새로운 연결이 추가되었습니다. 현재 연결 수: {}", emitters.size());
    sseEmitter.onTimeout(() -> {
        emitters.remove(sseEmitter);
        log.debug("[SSE] 연결이 타임아웃으로 종료되었습니다. 현재 연결 수: {}", emitters.size());
    });
    sseEmitter.onError(e -> {
        emitters.remove(sseEmitter);
    });
}

• emitters 전체를 조회하는 로직

// sseService.class
public void propagate(String eventName, Object data) {
    sseEmitterStorage.getEmitters().forEach(emitter -> {
        SseEventBuilder eventBuilder = SseEventBuilderFactory.create(eventName, data);
        sendEvent(emitter, eventBuilder);
    });
}

그래서 동시성을 보장해주는 ConcurrentLinkedQueue로 변경했어요. 하지만 LinkedQueue인 만큼 조회, 삭제 등에 속도가 느려요.. 추후 변경이 필요할 것으로 생각해요. HashMap은 어떨까 고민하고 있어요. 사용자마다 하나의 sse 커넥션을 맺는다는 점을 고려하여 key: token, value: sseEmitter로 두어서 관리하는 것이죠. 이렇게 하면 한 사용자는 반드시 하나의 연결만 맺도록 관리할 수 있을 것 같아요.

끊긴 SSE 커넥션에 데이터를 보내는 문제

SseEmitter에 설정한 timeout 시간이 지나면 sse 연결은 끊겨요! sse 연결이 끊기면 SseEmitter를 SseEmitterStorage에 저장할 때 등록한 콜백에 따라 스토리지에서 제거돼요.

public void add(SseEmitter sseEmitter) {
    emitters.add(sseEmitter);
    log.info("[SSE] 새로운 연결이 추가되었습니다. 현재 연결 수: {}", emitters.size());
    sseEmitter.onTimeout(() -> {
        emitters.remove(sseEmitter);
        log.debug("[SSE] 연결이 타임아웃으로 종료되었습니다. 현재 연결 수: {}", emitters.size());
    });
    sseEmitter.onError(e -> {
        emitters.remove(sseEmitter);
    });
}

문제는 SseEmitterStorage의 getEmitters에 있는데요.

public List<SseEmitter> getEmitters() {
    return emitters.stream().toList();
}

이때 위 코드와 같이 작성하면 getEmitters 메서드를 호출하는 시점에 가지고 있는 SseEmitter 목록을 그대로 전달해요. 갖고있는 참조를 그대로 보내지 않고 복사를 떠서 보내기 때문이에요. 관련해서는 이 글을 참고하면 좋아요. getEmitters 메서드 호출한 후에 timeout으로 인해 일부 SseEmitter 객체의 커넥션이 끊길 수 있어요. 하지만 스케줄러에서는 그것을 알지 못하고 데이터를 전송하려하고 예외가 발생해요.

이거는,,, 일단은 무시해도 치명적인 에러가 발생하지 않아서 무시하기로 했어요. 또한 timeout 시간을 길게 잡는다면 크게 문제가 안될 수 있어요.

SSE에서 에러가 발생하면 어디에서 핸들링하는 것이 좋은가

SSE 통신에서 문제가 생기면 예외를 잡을 수 있는 지점은 두 곳이 있어요.

1. SseEmitter의 콜백함수
2. 비즈니스 로직

저는 SseEmitter의 콜백함수에서 잡는 것이 좋다고 생각하는데요. SSE 연결을 다루는 근원이기 때문에 이것에서 핸들링하는 것이 예외 전파를 막을 수 있겠다 생각했어요. 하지만 지금은 그러지 않았는데요. 이유는,,, SseEmitter에서 에러처리해도 얻을 수 있는 정보가 매우 제한되더라고요

예시 상황은 위에서 본, 클라이언트가 일방적으로 연결을 끊은 상황이에요. 콜백함수에서 잡게 되면 다음과 같아요.

sseEmitter.onError(e -> {
    emitters.remove(sseEmitter);
    log.error("SSE 예외가 발생했습니다. {}", e);
});

단지 IOException만 발생하고 끝나더군요.. 관련 정보를 더 얻을 수 없어서 에러 핸들링이 어려웠어요.

대신에 비즈니스 로직에서 잡게 되면 다음과 같이 좀더 자세한 정보가 나와요.

private void sendEvent(SseEmitter sseEmitter, SseEventBuilder eventBuilder) {
    try {
        sseEmitter.send(eventBuilder);
    } catch (IOException e) {
        SseErrorHandler.handle(e);
    }
}

변경 사항 정리

SseStorage 구현체 변경

SseStorage 내부 구현체를 ConcurrentLinkedQueue에서 ConcurrentHashMap으로 변경했어요. ConcurrentLinkedQueue는 다음 단점이 있어요.

• 조회, 삭제 시간 복잡도가 O(N)이다.
• 조회, 삽입, 삭제 작업 모두가 빈번하게 일어나서, 동시성 처리에 오버헤드가 클 수 있다.

HashMap의 구현체인 ConcurrentHashMap으로 변경하면 예상되는 효과는 다음과 같아요.

• 조회, 삽입, 삭제 시간 복잡도가 O(1)이다.
• Map<사용자토큰, SseEmitter>로 관리한다면 사용자 별로 1개의 SSE 연결임을 보장할 수 있다.
• 사용자 별로 다른 데이터를 전송해야 하는 경우에, 사용자를 식별할 수 있다.

LocalDateTime.class가 직렬화 안되는 문제

ObjectMapper는 LocalDateTime.class를 제대로 직렬화하지 못하는데요.

com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Java 8 date/time type `java.time.LocalDateTime` not supported by default: add Module "com.fasterxml.jackson.datatype:jackson-datatype-jsr310" to enable handling at 
...

실제로 직렬화한 결과는 다음과 같아요.

// expected
{
    "queryTime":"2025-1-28T23:55:23.434294"
}

// actual (아마 이랬던 것 같아요)
{
    "queryTime": [
        "2025", "1", "28", "23", "55", "23", "434294"
    ]
}

따라서 ObjectMapper에 모듈을 추가하고, 또 하나의 설정을 해주었어요.