Server-Sent Events(SSE) 구현과 실무에서 마주할 수 있는 문제 해결기

@RestController
@RequestMapping("/api/sse")
public class SseController {
    
    private final SseEmitterService sseEmitterService;
    
    public SseController(SseEmitterService sseEmitterService) {
        this.sseEmitterService = sseEmitterService;
    }
    
    // SSE 연결을 위한 엔드포인트
    // produces 속성으로 text/event-stream Content-Type 설정
    @GetMapping(value = "/notifications", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribeToNotifications(@RequestParam String userId) {
        return sseEmitterService.createEmitter(userId);
    }
    
    // 알림 전송을 위한 엔드포인트
    @PostMapping("/send")
    public ResponseEntity<String> sendNotification(@RequestBody NotificationRequest request) {
        sseEmitterService.sendNotification(request.getUserId(), request.getMessage());
        return ResponseEntity.ok("Notification sent");
    }
}

@Service
@Slf4j
public class SseEmitterService {
    
    // 사용자 ID를 키로 하는 SSE 연결 관리
    // ConcurrentHashMap으로 스레드 안전성 보장
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    
    /**
     * 새로운 SSE 연결을 생성하고 관리
     * @param userId 사용자 식별자
     * @return SseEmitter 인스턴스
     */
    public SseEmitter createEmitter(String userId) {
        // 무제한 타임아웃으로 설정 (Long.MAX_VALUE)
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        
        // 연결 완료 시 처리 - 정상적인 연결 종료
        emitter.onCompletion(() -> {
            log.info("SSE connection completed for user: {}", userId);
            emitters.remove(userId); // 메모리 누수 방지
        });
        
        // 타임아웃 시 처리 - 네트워크 문제나 클라이언트 비정상 종료
        emitter.onTimeout(() -> {
            log.warn("SSE connection timeout for user: {}", userId);
            emitters.remove(userId);
        });
        
        // 에러 시 처리 - 예외 발생 시
        emitter.onError((ex) -> {
            log.error("SSE connection error for user: {}", userId, ex);
            emitters.remove(userId);
        });
        
        // 사용자별 연결 저장
        emitters.put(userId, emitter);
        
        // 초기 연결 확인 메시지 전송
        try {
            emitter.send(SseEmitter.event()
                .name("connect")  // 이벤트 타입 지정
                .data("Connected to SSE server")); // 실제 데이터
        } catch (IOException e) {
            log.error("Failed to send initial message", e);
            emitters.remove(userId);
        }
        
        return emitter;
    }
    
    /**
     * 특정 사용자에게 알림 메시지 전송
     * @param userId 대상 사용자 ID
     * @param message 전송할 메시지
     */
    public void sendNotification(String userId, String message) {
        SseEmitter emitter = emitters.get(userId);
        if (emitter != null) {
            try {
                // 구조화된 이벤트로 메시지 전송
                emitter.send(SseEmitter.event()
                    .name("notification")  // 이벤트 타입
                    .data(message));       // 메시지 내용
            } catch (IOException e) {
                log.error("Failed to send notification to user: {}", userId, e);
                emitters.remove(userId); // 연결이 끊어진 경우 제거
            }
        }
    }
    
    /**
     * 현재 활성 연결 수 반환 (모니터링용)
     * @return 활성 연결 수
     */
    public int getActiveConnectionCount() {
        return emitters.size();
    }
}

/**
 * SSE 클라이언트 클래스
 * 서버와의 SSE 연결을 관리하고 재연결 로직을 포함
 */
class SSEClient {
    constructor(userId) {
        this.userId = userId;
        this.eventSource = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 5;  // 최대 재연결 시도 횟수
        this.reconnectDelay = 1000;     // 초기 재연결 지연 시간 (ms)
    }
    
    /**
     * SSE 서버에 연결
     * EventSource를 생성하고 이벤트 리스너를 설정
     */
    connect() {
        try {
            // EventSource 생성 - 자동으로 GET 요청을 보내고 스트림 연결
            this.eventSource = new EventSource(`/api/sse/notifications?userId=${this.userId}`);
            
            // 연결 성공 시 호출
            this.eventSource.onopen = (event) => {
                console.log('SSE connection opened');
                this.reconnectAttempts = 0; // 연결 성공 시 재연결 카운터 리셋
            };
            
            // 기본 메시지 이벤트 처리 (event 타입이 지정되지 않은 경우)
            this.eventSource.onmessage = (event) => {
                console.log('Received message:', event.data);
                this.handleMessage(event.data);
            };
            
            // 커스텀 이벤트 리스너 - 'notification' 이벤트 처리
            this.eventSource.addEventListener('notification', (event) => {
                console.log('Received notification:', event.data);
                this.showNotification(event.data);
            });
            
            // 커스텀 이벤트 리스너 - 'connect' 이벤트 처리
            this.eventSource.addEventListener('connect', (event) => {
                console.log('Connected to SSE server');
            });
            
            // 에러 발생 시 호출
            this.eventSource.onerror = (event) => {
                console.error('SSE connection error:', event);
                this.handleError();
            };
            
        } catch (error) {
            console.error('Failed to create EventSource:', error);
            this.scheduleReconnect(); // 에러 발생 시 재연결 시도
        }
    }
    
    /**
     * 수신된 메시지 처리
     * @param {string} data 서버에서 전송된 데이터
     */
    handleMessage(data) {
        try {
            // JSON 형태의 데이터를 파싱 (필요한 경우)
            const notification = JSON.parse(data);
            this.showNotification(notification);
        } catch (e) {
            // JSON 파싱 실패 시 원본 데이터 그대로 사용
            this.showNotification(data);
        }
    }
    
    /**
     * 알림 표시
     * @param {string} message 표시할 메시지
     */
    showNotification(message) {
        // 브라우저 알림 권한이 있는 경우 시스템 알림 표시
        if ('Notification' in window && Notification.permission === 'granted') {
            new Notification('새 알림', { body: message });
        }
        
        // DOM 업데이트 (알림 UI 갱신)
        this.updateNotificationUI(message);
    }
    
    /**
     * 에러 처리 및 재연결 로직
     */
    handleError() {
        // 기존 연결 정리
        if (this.eventSource) {
            this.eventSource.close();
            this.eventSource = null;
        }
        
        // 최대 재연결 시도 횟수 내에서 재연결 시도
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.scheduleReconnect();
        } else {
            console.error('Max reconnection attempts reached');
            // 최대 시도 횟수 초과 시 사용자에게 알림
            this.showNotification('서버 연결이 끊어졌습니다. 페이지를 새로고침해주세요.');
        }
    }
    
    /**
     * 지수 백오프 전략을 사용한 재연결 스케줄링
     * 재연결 시도마다 대기 시간을 2배씩 증가시켜 서버 부하 방지
     */
    scheduleReconnect() {
        this.reconnectAttempts++;
        // 지수 백오프: 1초, 2초, 4초, 8초, 16초...
        const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
        
        console.log(`Scheduling reconnect in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
        
        setTimeout(() => {
            console.log(`Attempting to reconnect (${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
            this.connect();
        }, delay);
    }
    
    /**
     * 연결 종료
     * 페이지를 떠날 때 호출하여 리소스 정리
     */
    disconnect() {
        if (this.eventSource) {
            this.eventSource.close();
            this.eventSource = null;
        }
    }
}
 
// 사용 예시
const sseClient = new SSEClient('user123');
sseClient.connect();
 
// 페이지를 떠날 때 연결 정리
window.addEventListener('beforeunload', () => {
    sseClient.disconnect();
});

// 페이지를 떠날 때 SSE 연결 정리
// beforeunload 이벤트는 페이지 새로고침, 브라우저 닫기, 다른 페이지 이동 시 발생
window.addEventListener('beforeunload', () => {
    if (sseClient) {
        sseClient.disconnect(); // EventSource 연결 종료
    }
});

/**
 * 주기적으로 죽은 SSE 연결을 정리하는 스케줄러
 * 30초마다 실행되어 메모리 누수를 방지
 */
@Scheduled(fixedRate = 30000) // 30초마다 실행
public void cleanupDeadConnections() {
    // 모든 연결을 순회하면서 상태 확인
    emitters.entrySet().removeIf(entry -> {
        SseEmitter emitter = entry.getValue();
        try {
            // 핑 메시지 전송으로 연결 상태 확인
            // 성공하면 연결이 살아있는 것이므로 false 반환 (제거하지 않음)
            emitter.send(SseEmitter.event()
                .name("ping")
                .data("ping"));
            return false;
        } catch (IOException e) {
            // 핑 전송 실패 시 연결이 끊어진 것으로 판단
            log.info("Removing dead connection for user: {}", entry.getKey());
            return true; // true 반환하여 Map에서 제거
        }
    });
}

/**
 * 사용자별 SSE 연결 생성
 * 기존 연결이 있다면 정리 후 새로운 연결 생성
 * @param userId 사용자 식별자
 * @return 새로운 SseEmitter 인스턴스
 */
public SseEmitter createEmitter(String userId) {
    // 기존 연결이 있는지 확인
    SseEmitter existingEmitter = emitters.get(userId);
    if (existingEmitter != null) {
        try {
            // 기존 연결에 종료 메시지 전송
            // 클라이언트가 정상적으로 연결 종료를 인지할 수 있도록 함
            existingEmitter.send(SseEmitter.event()
                .name("disconnect")
                .data("New connection from another tab"));
            
            // 연결 완료 처리 (onCompletion 콜백 호출)
            existingEmitter.complete();
            
            log.info("Replaced existing connection for user: {}", userId);
        } catch (IOException e) {
            // 이미 끊어진 연결인 경우 무시
            log.warn("Failed to send disconnect message to existing connection", e);
        }
    }
    
    // 새로운 연결 생성 (기존 로직과 동일)
    SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
    
    // 생명주기 콜백 설정
    emitter.onCompletion(() -> {
        log.info("SSE connection completed for user: {}", userId);
        emitters.remove(userId);
    });
    
    emitter.onTimeout(() -> {
        log.warn("SSE connection timeout for user: {}", userId);
        emitters.remove(userId);
    });
    
    emitter.onError((ex) -> {
        log.error("SSE connection error for user: {}", userId, ex);
        emitters.remove(userId);
    });
    
    // 새로운 연결 저장
    emitters.put(userId, emitter);
    
    return emitter;
}

# SSE 엔드포인트에 대한 프록시 설정
location /api/sse/ {
    # 백엔드 서버로 요청 전달
    proxy_pass http://backend;
    
    # HTTP/1.1 사용 (keep-alive 지원)
    proxy_http_version 1.1;
    
    # Connection 헤더 제거 (keep-alive 연결 유지)
    proxy_set_header Connection "";
    
    # 클라이언트 정보 전달
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header Host $http_host;
    
    # SSE 연결 유지를 위한 핵심 설정
    proxy_buffering off;        # 응답 버퍼링 비활성화 (실시간 전송)
    proxy_cache off;            # 캐싱 비활성화 (항상 최신 데이터)
    proxy_read_timeout 86400s;  # 읽기 타임아웃 24시간
    proxy_send_timeout 86400s;  # 쓰기 타임아웃 24시간
}

# application.yml
server:
  tomcat:
    threads:
      max: 200  # 동시 처리 가능한 스레드 수 증가
    connection-timeout: 86400000  # 연결 타임아웃 24시간 (밀리초)
  
  # 추가적인 SSE 관련 설정
  servlet:
    async:
      request-timeout: 86400000  # 비동기 요청 타임아웃

{
  "TargetGroupAttributes": [
    {
      "Key": "deregistration_delay.timeout_seconds",
      "Value": "0"
    },
    {
      "Key": "stickiness.enabled",
      "Value": "true"
    },
    {
      "Key": "stickiness.lb_cookie.duration_seconds",
      "Value": "86400"
    }
  ]
}

@Service
public class SseEventQueue {
    
    private final Map<String, BlockingQueue<SseEvent>> eventQueues = new ConcurrentHashMap<>();
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    
    public void enqueueEvent(String userId, SseEvent event) {
        BlockingQueue<SseEvent> queue = eventQueues.computeIfAbsent(userId, 
            k -> new LinkedBlockingQueue<>());
        
        queue.offer(event);
        executorService.submit(() -> processEvents(userId));
    }
    
    private void processEvents(String userId) {
        BlockingQueue<SseEvent> queue = eventQueues.get(userId);
        SseEmitter emitter = sseEmitterService.getEmitter(userId);
        
        if (emitter == null) {
            eventQueues.remove(userId);
            return;
        }
        
        try {
            SseEvent event = queue.poll();
            if (event != null) {
                emitter.send(SseEmitter.event()
                    .name(event.getType())
                    .data(event.getData()));
            }
        } catch (IOException e) {
            log.error("Failed to send queued event", e);
            eventQueues.remove(userId);
        }
    }
}

class NotificationService {
    constructor(userId) {
        this.userId = userId;
        this.sseClient = null;
        this.pollingInterval = null;
    }
    
    start() {
        if (typeof EventSource !== 'undefined') {
            // SSE 지원하는 경우
            this.sseClient = new SSEClient(this.userId);
            this.sseClient.connect();
        } else {
            // SSE 지원하지 않는 경우 폴링 사용
            this.startPolling();
        }
    }
    
    startPolling() {
        this.pollingInterval = setInterval(() => {
            this.fetchNotifications();
        }, 5000); // 5초마다 폴링
    }
    
    async fetchNotifications() {
        try {
            const response = await fetch(`/api/notifications?userId=${this.userId}`);
            const notifications = await response.json();
            notifications.forEach(notification => {
                this.showNotification(notification);
            });
        } catch (error) {
            console.error('Failed to fetch notifications:', error);
        }
    }
    
    stop() {
        if (this.sseClient) {
            this.sseClient.disconnect();
        }
        if (this.pollingInterval) {
            clearInterval(this.pollingInterval);
        }
    }
}

@Component
@Slf4j
public class SseMetrics {
    
    private final MeterRegistry meterRegistry;
    private final SseEmitterService sseEmitterService;
    
    public SseMetrics(MeterRegistry meterRegistry, SseEmitterService sseEmitterService) {
        this.meterRegistry = meterRegistry;
        this.sseEmitterService = sseEmitterService;
    }
    
    @Scheduled(fixedRate = 10000) // 10초마다 실행
    public void recordMetrics() {
        int activeConnections = sseEmitterService.getActiveConnectionCount();
        meterRegistry.gauge("sse.active.connections", activeConnections);
        
        log.info("Active SSE connections: {}", activeConnections);
    }
}

@Aspect
@Component
public class SsePerformanceAspect {
    
    private final MeterRegistry meterRegistry;
    
    public SsePerformanceAspect(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    
    @Around("execution(* com.example.service.SseEmitterService.sendNotification(..))")
    public Object measureSendTime(ProceedingJoinPoint joinPoint) throws Throwable {
        Timer.Sample sample = Timer.start(meterRegistry);
        
        try {
            return joinPoint.proceed();
        } finally {
            sample.stop(Timer.builder("sse.send.duration")
                .tag("method", "sendNotification")
                .register(meterRegistry));
        }
    }
}

@Service
public class SseConnectionPool {
    
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    private final int maxConnections;
    private final AtomicInteger currentConnections = new AtomicInteger(0);
    
    public SseConnectionPool(@Value("${sse.max-connections:1000}") int maxConnections) {
        this.maxConnections = maxConnections;
    }
    
    public SseEmitter createEmitter(String userId) {
        if (currentConnections.get() >= maxConnections) {
            throw new RuntimeException("Maximum SSE connections reached");
        }
        
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        currentConnections.incrementAndGet();
        
        emitter.onCompletion(() -> {
            emitters.remove(userId);
            currentConnections.decrementAndGet();
        });
        
        emitter.onTimeout(() -> {
            emitters.remove(userId);
            currentConnections.decrementAndGet();
        });
        
        emitter.onError((ex) -> {
            emitters.remove(userId);
            currentConnections.decrementAndGet();
        });
        
        emitters.put(userId, emitter);
        return emitter;
    }
}

@RestController
@RequestMapping("/api/sse")
public class SecureSseController {
    
    @GetMapping(value = "/notifications", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribeToNotifications(
            @RequestParam String userId,
            @AuthenticationPrincipal UserDetails userDetails) {
        
        // 사용자 권한 확인
        if (!userDetails.getUsername().equals(userId)) {
            throw new AccessDeniedException("Access denied");
        }
        
        return sseEmitterService.createEmitter(userId);
    }
}

@Configuration
public class WebConfig implements WebMvcConfigurer {
    
    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/api/sse/**")
            .allowedOrigins("https://yourdomain.com")
            .allowedMethods("GET")
            .allowCredentials(true);
    }
}