SSE를 이용한 실시간 알람 기능 구현에 대해 글을 작성해 보고자 한다. 기본적인 구현 방법과 내가 직접 작성한 코드를 통해 어떻게 구현 했는지 포스팅 해보려고 한다.
관련 이론은 다른 글에 포스팅해 놓았다. 여기를 클릭하면 확인할 수 있다.
구현 환경 : Spring boot + Next.js
개요
SSE(Server-Sent Events)는 서버에서 클라이언트로 데이터를 지속적으로 전송할 수 있는 단방향 실시간 통신 기술이다. 클라이언트는 EventSource 객체를 이용하여 서버에 구독 요청을 보내고, 서버는 text/event-stream 형식으로 연결을 유지한 채 데이터를 지속적으로 전달한다. 아래 그림은 SSE의 기본 동작 흐름이다.
SSE 네트워크 프로토콜
연결(구독, Subscribe)
Client: SSE Subscrib 요청
(위 그림의 Get /connect 에 해당하는 부분이다.)
GET /sse HTTP/1.1
Accept: text/event-stream // 이벤트 미디어 타입의 표준은 'text/event-stream'
Cache-Control: no-cache // 이벤트 캐싱 없음
Connection: keep-alive // 지속적 연결 사용
Server: Subscription에 대한 응답
(위 그림의 text/event-stream을 EventSource로 응답하는 부분이다.)
HTTP/1.1 200
Content-Type: text/event-stream;charset=UTF-8 // 응답 미티어 타입은 text/event-stream
Transfer-Encoding: chunked // 서버는 동적으로 생성된 컨텐츠를 스트리밍하기 때문에 분문의 크기를 미리 알 수 없음
데이터 전송
Server: 이벤트 전달
(위 그림의 data change! 로 응답하는 클라이언트에게 데이터를 전달해 주는 부분이다.)
구독(Subscribe)이 완료되면 서버는 새로운 데이터가 생길 때마다 클라이언트에게 메시지를 전송한다. SSE 이벤트는 UTF-8 인코딩 기반의 텍스트 형식으로 전달된다.
각 이벤트(Event)는 \n\n(줄바꿈 2번) 으로 구분된다. 그리고 하나의 이벤트 내부에는 다음과 같은 형식의 필드가 들어갈 수 있다.
name: value
// 각 이벤트(Event)는 \n\n(줄바꿈 2번) 으로 구분된다.
// 각 필드는 \n(줄바꿈 1번)으로 구분된다.
//id 필드
id: 1
data: The first event.
id: 2
data: The second event.
//Event 필드
event: type1
data: An event of type1.
event: type2
data: An event of type2.
흐름
1. 클라이언트에서 SSE 연결 요청
2. 서버에서는 요청한 브라우저에 해당하는 SSE 통신 객체 생성
3. 향후 서버에서 이벤트가 발생하면 해당 객체를 통해 클라이언트로 데이터 전달
구현
아키텍쳐
프론트 엔드
1. 브라우저에서 SSE 연결 요청하기
프론트엔드에서는 브라우저가 제공하는 EventSource 객체를 사용하여 서버에 SSE 연결을 요청한다.
const es = new EventSource(
`${process.env.NEXT_PUBLIC_API_BASE}/api/notifications/stream`,
{ withCredentials: true }
);
// ${process.env.NEXT_PUBLIC_API_BASE} 부분은 ip주소이다.
// withCredentials: true를 설정한 이유는 로그인 인증에 사용되는 쿠키를 함께 전송하기 위해서이다.EventSource는 SSE 연결을 생성하는 브라우저 내장 객체이다. 위 코드는 백엔드의 /api/notifications/stream 엔드포인트로 연결 요청을 보내고, 서버는 이 요청에 대해 text/event-stream 형식으로 응답한다.
2. 서버에서 보낸 notification 이벤트 수신하기
서버에서는 알림 데이터를 notification이라는 이벤트 이름으로 전송한다. 따라서 프론트엔드에서는 addEventListener()를 이용해 해당 이벤트를 수신한다.
const handle = (ev: MessageEvent) => {
const data = JSON.parse(ev.data) as NotificationDTO;
onMessage(data);
};
es.addEventListener("notification", handle as EventListener);
서버에서 다음과 같은 SSE 메시지를 보내면,
event: notification
data: {"id":1,"message":"새 댓글이 작성되었습니다."}브라우저에서는 notification 이벤트 리스너가 실행된다.
3. 컴포넌트 종료 시 SSE 연결 닫기
React 컴포넌트가 사라질 때는 기존 SSE 연결을 정리해야 한다.
return () => {
es.removeEventListener("notification", handle as EventListener);
es.close();
};es.close()를 호출하지 않으면 사용자가 페이지를 이동해도 기존 연결이 남아 있을 수 있다. 따라서 useEffect의 cleanup 함수에서 이벤트 리스너를 제거하고 SSE 연결을 닫아준다.
4. 프론트엔드 전체 코드
"use client";
import { useEffect } from "react";
import { NotificationDTO } from "../myPage/types/Notification";
export function useNotificationStream(
onMessage: (n: NotificationDTO) => void
) {
useEffect(() => {
const es = new EventSource(
`${process.env.NEXT_PUBLIC_API_BASE}/api/notifications/stream`,
{ withCredentials: true }
);
const handle = (ev: MessageEvent) => {
try {
const data = JSON.parse(ev.data) as NotificationDTO;
onMessage(data);
} catch {
console.warn("알림 데이터 파싱 실패:", ev.data);
}
};
es.addEventListener("notification", handle as EventListener);
es.onerror = () => {
// 브라우저가 자동으로 재연결을 시도한다.
};
return () => {
es.removeEventListener("notification", handle as EventListener);
es.close();
};
}, [onMessage]);
}
백엔드
1. SSE 연결 요청 처리
클라이언트가 /stream 엔드포인트로 연결 요청을 보내면 서버는 SseEmitter 객체를 생성하여 SSE 연결을 유지한다.
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) //TEXT_EVENT_STREAM_VALUE는 SSE 전용 MIME 타입
public SseEmitter stream(HttpServletRequest req) { // 클라이언트가 EventSource로 연결 요청 시 실행 된다.
Long userId = extractUserIdFromRequest(req); // JWT 기반 로그인 사용자 정보를 추출
return streamer.subscribe(userId); // NotificationStreamer 실제 SSE 연결 생성
}
2. SseEmitter 생성 및 연결 관리
실제 SSE 연결 관리는 NotificationStreamer 클래스에서 처리하였다.
private final ConcurrentMap<Long, CopyOnWriteArrayList<SseEmitter>> emitters
= new ConcurrentHashMap<>();
알림은 특정 사용자에게 전달되어야 하기 때문에 Map 형태로 관리했다.
userId -> emitter 목록
3. SSE 구독(Subscribe)
클라이언트가 연결 요청을 보내면 서버는 새로운 SseEmitter 객체를 생성한다.
public SseEmitter subscribe(Long userId) {
SseEmitter emitter = new SseEmitter(0L);
emitters.computeIfAbsent(
userId,
k -> new CopyOnWriteArrayList<>())
.add(emitter);
return emitter;
}
4. 연결 종료 처리
브라우저 종료, 네트워크 문제 등으로 SSE 연결이 끊어질 수 있기 때문에 종료된 emitter를 제거해야 한다.
emitter.onCompletion(() -> {
remove(userId, emitter);
});
emitter.onTimeout(() -> {
remove(userId, emitter);
});
emitter.onError(e -> {
remove(userId, emitter);
});연결이 종료된 emitter를 제거하지 않으면 메모리에 계속 남게 된다. 이는 메모리 누수(Memory Leak) 문제로 이어질 수 있다.
5. 실시간 알림 전송
새로운 알림이 생성되면 저장된 SseEmitter를 통해 실시간 데이터를 전송한다.
public void push(NotificationEntity n) {
var list = emitters.getOrDefault(
n.getRecipientId(),
new CopyOnWriteArrayList<>());
for (SseEmitter e : list) {
e.send(
SseEmitter.event()
.name("notification")
.id(String.valueOf(n.getId()))
.data(NotificationDTO.from(n))
);
}
}
실제 전송되는 SSE 형식
위 코드는 실제로 다음과 같은 SSE 메시지 형태로 전송된다.
event: notification
id: 15
data: {"message":"새 댓글이 작성되었습니다."}
5.백엔드 전체 코드
public class NotificationStreamer {
// 유저별 다중 탭 지원을 위해 List 보관
private final ConcurrentMap<Long, CopyOnWriteArrayList<SseEmitter>> emitters = new ConcurrentHashMap<>();
/** 유저가 구독할 때마다 Emitter 발급 */
public SseEmitter subscribe(Long userId) {
// 무한 타임아웃 (전역 설정도 함께 권장: spring.mvc.async.request-timeout=0)
SseEmitter emitter = new SseEmitter(0L);
emitters.computeIfAbsent(
userId,
k -> new CopyOnWriteArrayList<>())
.add(emitter);
log.info("SSE 연결 생성: userId={}", userId);
emitter.onCompletion(() -> {
log.info("SSE 연결 정상 종료: userId={}", userId);
remove(userId, emitter);
});
emitter.onTimeout(() -> {
log.warn("SSE 연결 타임아웃: userId={}", userId);
remove(userId, emitter);
});
emitter.onError(e -> {
log.warn("SSE 연결 오류: userId={}, error={}", userId, e.getMessage());
remove(userId, emitter);
});
// 연결 확인용 이벤트 한 번 보내기
try {
emitter.send(SseEmitter.event().name("hello").data("ok"));
log.info("SSE hello 이벤트 전송 성공: userId={}", userId);
} catch (IOException e) {
log.warn("SSE hello 이벤트 전송 실패: userId={}, error={}", userId, e.getMessage());
remove(userId, emitter);
}
return emitter;
}
/** 저장된 알림을 실시간으로 해당 유저에게 브로드캐스트 */
public void push(NotificationEntity n) {
var list = emitters.getOrDefault(n.getRecipientId(), new CopyOnWriteArrayList<>());
for (SseEmitter e : list) {
try {
e.send(SseEmitter.event()
.name("notification")
.id(String.valueOf(n.getId()))
.data(NotificationDTO.from(n)));
} catch (IOException ex) {
remove(n.getRecipientId(), e); // 끊긴 연결 정리
}
}
}
코드리뷰
1. SseEmitter 생성 및 연결 관리 관련 코드
private final ConcurrentMap<Long, CopyOnWriteArrayList<SseEmitter>> emitters
= new ConcurrentHashMap<>();- 하나의 사용자가 PC 브라우저, 모바일 브라우저, 여러 탭 등 동시에 접속 할 수 있기 때문에 List 구조를 사용 (유저 아이디 Long 타입, 에미터 발급을 리스트 형태로 저장)
- ConcurrentHashMap : SSE 연결은 여러 사용자가 동시에 요청할 수 있기 때문에 멀티스레드 환경에서 안전한 자료구조가 필요하다. 따라서 thread-safe 한 ConcurrentHashMap을 사용하였다.
- CopyOnWriteArrayList : SSE 연결 중에는 emitter 추가, 제거, 순회(send) 와 같은 작업이 동시에 발생할 수 있다. 일반 ArrayList 사용 시 ConcurrentModificationException 문제가 발생할 수 있기 때문에 thread-safe 한 CopyOnWriteArrayList를 사용하였다.
질문이나 잘못된 내용 댓글남겨 주세요.
감사합니다.
참고 자료
https://github.com/aliakh/demo-spring-sse
https://gong-check.github.io/dev-blog/BE/%EC%96%B4%EC%8D%B8%EC%98%A4/sse/sse/