문제 상황
기본 개념
Emitter 관리 데이터 : Map<Long, Map<String, SseEmitter>>
=> <UserId, <emitterId, SseEmitter>>
emitterId : 한 사용자가 가진 여러 통로 고유ID (크롬, 모바일, 사파리 ...)
SseEmitter : 데이터를 보내는 HTTP 연결 객체 (통로)
문제
기본 코드에서는 UserId가 있다면 해당 사용자의 여러 고유ID중 특정 emitterId만 제거한 후, 남은 연결이 없다면 사용자의 미수신 이벤트 캐시(미수신 데이트)도 모두 정리했음. 이건 재시도 과정을 놓치고 설계한 설계실수.
-
스레드 A (삭제): 사용자가 마지막 탭을 닫음. isEmpty()를 확인하니 true임. "이제 userId를 지워야지!"라고 결심함.
-
스레드 B (추가): 그 찰나에 사용자가 새로운 탭을 열어서 userId 밑에 새 이메터를 추가함.
-
스레드 A (삭제): 아까 true였던 기억만 가지고 emitters.remove(userId)를 실행함.
-
결과: 스레드 B가 방금 만든 따끈따끈한 새 연결까지 같이 증발함. 사용자는 접속하자마자 알림을 못 받는 상태가 됨.
확인 후 실행 하고 있음.
즉, 연산 사이사이에 타 스레드 침범 가능성이 있음.
해결책 (Locking)
Emitter의 computeIfPresent() 사용.
: 해당 키가 속한 버킷(위의 데이터)에 락을 검. 하여 해당 로직이 끝날 때까지 다른 스레드는 이 userId를 건들 수 없음. 즉, 중간에 데이터가 바뀌지 않음
-> 원자성 보장
-> 일관성 보장
추가 학습
Q1. 만약 이 emitters 맵을 ConcurrentHashMap이 아닌 일반 HashMap으로 선언하고 멀티스레드 환경에서 운영한다면, 동시에 여러 사용자가 접속하거나 나갈 때 어떤 치명적인 오류가 발생할 수 있을까요? (힌트: ConcurrentModificationException)
: ConcrurrentHashMap은 스레드가 동시에 데이터를 넣고 빼도 데이터가 망가지지 않도록 설계된 자료구조 - 현재 SSE 연결할 때 이 구조 사용
[ 스레드 A가 1번 구역을 쓰는 동안, 스레드 B는 2번 구역을 동시에 쓸 수 있음 -> 맵 전체에 Lock을 거는 것보다 훨씬 빠르고 효율적임 ]
하지만 HashMap을 이용하였을 때,
1. 스레드 A가 맵을 돌면서 알림을 보내고 있는데 스레드B가 들어와서 알림을 추가하거나 삭제하면 에러가 남
2. 동시에 같은 위치의 자리에 데이터를 저장할 가능성이 있음
Q2. SseEmitter는 영원히 유지될까요? 만약 네트워크 문제로 이 '빨대'가 막히거나 서버가 설정한 '타임아웃' 시간이 지나면 어떤 일이 벌어질까요? 이때 우리가 공부한 remove 로직이 어떻게 호출되는지 설명해 보세요.
: 영원히 유지되지 않으며, 타임아웃이나 네트워크 오류 발생 시 스프링이 미리 등록된 콜백을 실행하여 remove로직을 자동으로 호출함
종료 사유
1. 정상 종료 : 모든 데이터를 보내고complete()를 호출
2. 타임 아웃 : 설정된 시간 동안 아무런 통신이 없을 때
3. 네트워크 오류 : 사용자가 브라우저 탭을 강제로 닫거나 인터넷 연결이 끊겼을 때
SseEmitter를 생성하는 시점에 문제가 생기면 아래 행동들을 하도록 콜백 등록을 해둠.
public SseEmitter subscribe(Long userId, String lastEventId) {
SseEmitter emitter = new SseEmitter(SSE_TIMEOUT);
String emitterId = sseEmitterRepository.save(userId, emitter);
// 작업이 끝났을 때
emitter.onCompletion(() -> sseEmitterRepository.remove(userId, emitterId));
// 시간이 다 되었을 때
emitter.onTimeout(() -> sseEmitterRepository.remove(userId, emitterId));
// 에러 발생 (네트워크 끊김 ,, )
emitter.onError(e -> {
log.warn("SSE emitter error userId={}, emitterId={}: {}", userId, emitterId, e.getMessage());
sseEmitterRepository.remove(userId, emitterId);
});
try {
emitter.send(SseEmitter.event()
.name("connect")
.data("connected"));
} catch (IOException e) {
sseEmitterRepository.remove(userId, emitterId);
throw new CustomException(ErrorCode.SSE_CONNECTION_ERROR);
}
if (lastEventId != null && !lastEventId.isBlank()) {
List<SseEmitterRepository.CachedEvent> missedEvents =
sseEmitterRepository.getEventsAfter(userId, lastEventId);
for (SseEmitterRepository.CachedEvent event : missedEvents) {
try {
emitter.send(SseEmitter.event()
.id(event.eventId())
.name("notification")
.data(event.data()));
} catch (IOException e) {
sseEmitterRepository.remove(userId, emitterId);
break;
}
}
}
return emitter;
}Q3. 만약 사용자가 수백만 명이고 알림이 초당 수천 건씩 발생하는 대규모 시스템이라면, 이렇게 키마다 락을 거는 방식이 서버 성능에 어떤 영향을 줄 수 있을까요? 그리고 이를 개선하기 위해 'Redis' 같은 외부 저장소를 도입한다면 어떤 이점이 있을까요?
: 키 단위 락은 연산 자체보다 메모리 점유 측면에서 한계가 있음.
1. 락 경합 : 수만 개의 스레드가 서로 다른 userId를 수정할 때는 영향이 없음. 하지만 특정 유저에게 수만 건의 알림이 몰려 동일한 키에 접근하는 경우, 스레드들이 줄을 서게 되어 CPU 대기 시간 급증
=> ConcurrentHashMap은 내부적으로 데이터를 여러 개의 버킷 구역에 나눠 담음. 스레드 100개가 각각 서로 다른 userID를 수정한다면, userId들은 해시 함수에 의해 서로 다른 버킷에 배정될 학률이 높음.
하지만 동인 userId로 접근할 경우 동일한 버킷에 저장될것이기에 큐가 길어짐.
- 메모리 부하, GC 지연 : SseEmitter 객체를 JVM 힙 메모리에 들고 있으면, GC가 메모리를 청소할 때마다 훑어야 할 객체가 너무 많아져 서비스가 순간적으로 멈추는 현상이 길어짐
Redis 도입 시 이점
1. 분산 환경 지원
- 어느 서버에서 이벤트가 터지든 Redis에 메시지 발행 (kafka가 할 수도 있지. kafka가 더 안정적)
- 모든 서버는 Redis를 구독하고 있다가, 자기 서버에 A 유저가 연결되어 있다면 즉시 알림을 보냄. (실시간성 보장)
2. 이벤트 캐싱 및 복구
- 서버 메모리가 아닌 Redis에 미수신 알림을 저장하면, 일시적으로 네트워크가 끊겼다 재접속해도 Redis에서 누락된 데이터를 찾아 복구 가능
