알림 기능을 구현하면서 SSE에 대해 학습하게 되었다.
먼저 SSE가 무엇인지 알아보자.
SSE : Server-Sent-Event의 약자로, 서버에서 클라이언트로 실시간 이벤트를 전달하는 웹 기술.
배경
HTTP 특징인 비연결성은 연결한 적이 있어도 연결을 끊어버린다는 것이다.
이를 해결하기 위한 웹 기술은 Polling, Long Polling, WebSocket 그리고 SSE가 있다.
여기서 SSE는 단방향 통신이며 클라이언트의 별도 추가요청 없이 서버에서 업데이트를 스트리밍할 수 있다는 특징을 가진다.
장점과 단점
장점
1. HTTP를 통해 통신하므로 다른 프로토콜은 필요가 없고, 구현이 굉장히 쉽다는 것이다.
2. 네트워크 연결이 끊겼을 때 자동으로 재연결을 시도한다.
3. 실시간으로 서버에서 클라이언트로 데이터를 전송할 수 있다. 폴링 같은 경우는 실시간이라고 보기 어려운 점이 있는데, 이러한 한계를 극복한다.
단점
1. GET 메소드만 지원하고, 파라미터를 보내는데 한계가 있다.
2. 단방향 통신이며, 한 번 보내면 취소가 불가능하다는 단점이 있다.
3. 클라이언트가 페이지를 닫아도 서버에서 감지하기가 어렵다는것도 단점이다.
4. SSE는 지속적인 연결을 유지해야 하므로, 많은 클라이언트가 동시에 연결을 유지할 경우 서버 부담이 커질 수 있다.
이제는 Spring에서 SSE를 구현해 볼 것이다.
SSE를 구현하기에 앞서 먼저 알아야 할 개념은 타임아웃이다.
타임아웃이란, 클라이언트 측에서 일정 시간 동안 서버로부터 데이터를 받지 못할 경우에 발생하는 상황을 말한다. 타임아웃이 발생하면 브라우저에서 자동으로 서버에 재연결 요청을 보내서 해결하게 된다.
SSE의 실행 과정
- 클라이언트가 서버의 이벤트를 구독하기 위한 요청을 보낸다.
- 서버에서는 클라이언트와 매핑되는 SSE객체를 만든다.
- 서버는 이벤트 스트림을 생성하고 클라이언트에게 비동기적으로 데이터를 전송한다.
1. Service에 SSE Emitter를 생성하고, 타임아웃을 설정 해 준다.
public class NotificationService {
//타임아웃 설정
private static final Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
private final NotificationRepository notificationRepository;
//SSE Emitter를 생성하는 메소드
private SseEmitter createEmitter(Long id) {
SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
//생성된 SSE Emitter를 저장소에 저장
notificationRepository.save(id, emitter);
// Emitter가 완료될 때(모든 데이터가 성공적으로 전송된 상태) Emitter를 삭제한다.
emitter.onCompletion(() -> notificationRepository.deleteById(id));
// Emitter가 타임아웃 되었을 때(지정된 시간동안 어떠한 이벤트도 전송되지 않았을 때) Emitter를 삭제한다.
emitter.onTimeout(() -> notificationRepository.deleteById(id));
return emitter;
}
}타임아웃을 1시간으로 설정했다. SSE Emitter를 생성하는 메소드를 작성하였고, Repo에 저장까지 하였다.
여기서 SSE 특성상, 한 번 데이터를 전송한 SSE Emitter는 단방향 통신을 위한 일회성 객체로 간주되어, 한 번 사용하고 나면 폐기되고 다시 생성해야 한다. 전송 중 오류가 발생하거나, 타임 아웃이 발생했을 경우에도 마찬가지이다.
2. 클라이언트에게 데이터를 전송하는 메소드를 설정한다.
//데이터를 클라이언트에게 보내는 메소드
private void sendEvent(Long sendId, Object data) {
// 먼저 클라이언트의 SseEmitter를 가져온다
SseEmitter emitter = notificationRepository.get(sendId);
if (emitter != null) {
try {
// 데이터를 클라이언트에게 실어보낸다.
emitter.send(SseEmitter.event().id(String.valueOf(sendId)).name("업무수정").data(data));
} catch (IOException exception) {
// 데이터 전송 중 오류가 발생하면 Emitter를 삭제하고 에러를 완료 상태로 처리
notificationRepository.deleteById(sendId);
emitter.completeWithError(exception);
}
}
}Repo에서 해당 Id에 할당된 SseEmitter를 가져온다. Emitter가 존재하면, "업무수정"이라는 데이터를 실어서 전송되게 끔 만들었다.
위에서 말했듯이 데이터 전송 중 오류가 발생하게 되면 Emitter를 삭제하고 다시 만들어야 한다.
3. 클라이언트가 구독을 호출하는 메소드를 만든다.
이는 클라이언트 Controller에서 구독페이지 엔드포인트를 생성하는데 사용된다.public SseEmitter subscribe(Long userId,final HttpServletResponse response) {
SseEmitter emitter = createEmitter(userId);
response.setContentType("text/event-stream");
response.setCharacterEncoding("UTF-8");
sendEvent(userId, "더미데이터" + userId + "]");
return emitter;
}userId를 이용하여 클라이언트와 매핑되는 Emitter를 생성해준다.
여기서, sendEvent를 이용하여 데이터를 보내주는 이유는 클라이언트와의 초기 연결에서 아무 이벤트도 전송하지 않으면, 재연결 요청이나 연결 자체에서 오류가 발생할 수 있다. 따라서 첫 SSE 응답을 보낼 시 더미 데이터를 넣어 이러한 오류를 방지한다.
4. 구독 페이지를 만든다.
private final NotificationService notificationService;
//구독 페이지
@GetMapping(value = "/subscribe/{id}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribe(@PathVariable Long id) {
return notificationService.subscribe(id);
}
@PostMapping("/send-data/{id}")
public void sendData(@PathVariable Long id) {
notificationService.notify(id, "data");
}
}
" /notifications/subscribe/{id} " 경로로 GET 요청이 오면, NotificationService를 사용하여 해당 ID에 대한 SSEEmitter를 생성하고 반환한다. 즉, 서버에서 데이터를 받게 된다.
또한, 해당 경로로 POST 요청이 오면, NotificationService를 사용하여 해당 ID에 대한 데이터를 클라이언트에게 전송한다.
5. Repository를 생성해준다.
@Repository
@RequiredArgsConstructor
public class NotificationRepository {
// 모든 Emitters를 저장하는 ConcurrentHashMap -> 여기서 이걸 쓴 이유 .
private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
//Emitters 저장
public void save(Long id, SseEmitter emitter) {
emitters.put(id, emitter);
}
//Emitter 제거
public void deleteById(Long id) {
emitters.remove(id);
}
//Emitter 가져오기
public SseEmitter get(Long id) {
return emitters.get(id);
}
}- Emitter 저장
클라이언트에게 보낼 id와 데이터를 저장한다. - Emitter 가져오기
정보를 담기위해서 클라이언트(id) 의 Emitter 를 가져와야한다. - Emitter 제거
Emitter가 완료될 때, 타임아웃 되었을 때, 오류가 발생했을 때 사용된다.
이로써 Spring 으로 SSE 구현이 끝이난다. 여기서 ConcurrentHashMap를 사용한 이유는 thread-safe 한 자료구조이기 때문이다.
자세한 내용은 아래서 다루겠다.
SSE 사용시 주의할 점
-
Emitter를 생성한 후 만료 시간까지 아무 데이터도 보내지 않으면, 재연결 요청 시 503 Service Unavailable 에러가 발생할 수 있다. 이러한 상황을 방지하기 위해 초기 SSE 연결 시 더미 데이터를 전송하여 안전한 연결을 유지한다.
-
thread-safe한 구조를 사용하지 않으면 ConcurrnetModificationException이 발생할 수 있다. 타임아웃 발생 시 실행할 콜백이 SseEmitter를 관리하는 다른 스레드에서 실행되기 때문이다. CopyOnWriteArrayList를 사용할 수도 있다.
-
JPA를 사용하는 동안 open-in-view 속성을 true로 설정하면 DB Connection Pool에서 동시에 많은 클라이언트가 SSE 연결을 시도할 경우 DB 커넥션 고갈이 발생할 수 있다. 이를 방지하기 위해 SSE 연결 동안에는 open-in-view 속성을 false로 설정하여 HTTP Connection이 닫힐 때마다 DB Connection도 적절히 해제되도록 한다.
조금만 더 탐구를 해보자면..
Connection Pool이란?
: DB와 연결해놓은 객체들을 Pool에 저장해두었다가 클라이언트 요청이 오면 커넥션을 빌려주고 다시 반납받아 Pool에 저장하는 방식을 말한다.
이것을 사용하면, 어플리케이션이 요청을 처리할 때마다 매번 데이터베이스에 연결을 맺지 않고, 이미 활성화된 연결을 재사용하여 응답 시간이 감소하고 리소스 효율성이 증가한다.
open-in-view 란?
: 웹 애플리케이션에서 데이터베이스와 상호작용할 때 트랜잭션을 요청-응답 주기와 일치시키는 방법을 말한다.
마지막으로 thread-safe 란 다음 장에서 알아보자.
마치며
이벤트가 발생하면 실시간으로 언제든 알림이 오게 끔 만들어야하는데, 타임아웃으로 연결이 끊어지면 이를 어떻게 처리해야 하는지에 대한 궁금증이 있었다. 이는 브라우저에서 자동으로 서버에 재연결 요청을 보내서 해결됨을 알 수 있다.
다른 방법으로는 타임아웃을 무제한으로 설정하는 방법이 있다. 이러면 Connection을 계속 유지할 수 있게된다. 그러나 이 방식은 서버 리소스를 계속 사용하며, 클라이언트 수가 증가할수록 서버 부하도 증가할 가능성이 생겼다. 또한 무한한 연결 시간 설정은 보안 문제와 무단 접근 가능성을 높일 수 있음을 알게되었다.
또 하나의 방법은 WebSocket을 이용하는 방법이다.
WebSocket에 대해 공부하기로 한다.
참고자료
https://dkswnkk.tistory.com/702
https://tecoble.techcourse.co.kr/post/2022-10-11-server-sent-events/
