SSE로 알림 기능 구현하기 with Spring

서론

인터넷은 웹 브라우저와 웹 서버 간의 데이터 통신을 위해서 HTTP 표준 위에 구축되어 있습니다. 대부분의 경우 웹 브라우저인 클라이언트가 HTTP 요청을 서버에 보내고, 서버는 적절한 응답을 반환하는데 이런 왕복 통신은 'https://www.google.com'과 같은 주소를 브라우저에 입력했을 때 웹 페이지를 받게 되는 과정입니다.

이러한 HTTP 표준은 광범위하게 지원되지만, 애플리케이션이 연속적인 정보를 서버에 전송하거나, 실시간으로 업데이트된 서버의 정보를 클라이언트에게 보내야 하는 경우 지속적인 HTTP 요청을 하게 되기에 비용면에서 매우 비 효율적입니다. 이런 상황에서 폴링, 웹소켓, 그리고 SSE가 등장했는데, 이들은 데이터 스트림의 속도와 메모리 효율성에 중점을 둔 프로토콜 들입니다.

1. Short Polling
2. Long Polling
3. SSE(Server-Sent-Events)
4. WebSocket

이번 게시글에서는 위 프로토콜들의 개념에 대해서 간략하게 설명하고, Spring에서 SSE를 활용해 알림 기능을 구현하는 예시를 작성해 보도록 하겠습니다.
 

1. Short Polling 

Short Polling은 클라이언트가 서버로부터 정기적인 정보를 받기 위한 프로토콜입니다. 클라이언트가 주기적으로 서버로 요청을 보내 데이터를 받아오는 방법입니다.

Short Polling

과정은 다음과 같습니다.

• 클라이언트가 서버에 새로운 정보에 대한 HTTP 요청을 보낸다.
• 서버는 새로운 정보를 클라이언트에게 반환한다.
• 클라이언트는 설정한 주기(예:2초)로 요청을 반복한다.

클라이언트가 Http Request를 서버로 계속 보내서 이벤트 내용을 전달받는 방식이며, 클라이언트가 지속적으로 Request를 서버에 보내기 때문에 클라이언트가 많아지면 서버의 부담이 증가하게 됩니다. 또한 TCP의 connection을 맺고 끊는 것 자체가 매우 무겁고, 이에 따라 실시간으로 변화되는 빠른 정보의 응답을 기대하기는 어렵습니다.

하지만 클라이언트와 서버의 구현이 모두 단순하다는 장점이 있고, 서버가 요청에 대한 부담이 크지 않고 요청 주기를 넉넉하게 잡아도 될 정도로 실시간성이 중요하지 않다면 고려해 볼 만한 방법입니다.
 

2. Long Polling

Long Polling은 Short Polling 보다는 더 효율적인 방식입니다.

Long Polling

Short Polling 보다 서버에서 기다리는 시간이 더 길어진 것이 보이시나요? 과정은 Short Polling에서 마지막만 다릅니다.

• 클라이언트가 서버에 새로운 정보에 대한 HTTP 요청을 보낸다.
• 서버는 새로운 정보를 클라이언트에게 반환한다.
• 클라이언트는 이전 응답을 받자마자 다시 요청을 반복한다.

Long Polling은 Short Polling에 비해 동일한 양의 데이터를 클라이언트에게 전송하는데 필요한 HTTP 요청 수를 줄일 수 있습니다. 하지만 서버로부터 응답을 받고 나면 다시 연결 요청을 하기 때문에, 상태가 빈번하게 바뀐다면 연결 요청도 늘어나 서버에 부담이 가는 것은 변하지 않습니다.

따라서 실시간 메시지 전달이 중요하지만 서버의 상태가 빈번하게 변하지 않는 경우에 적합합니다.
 

3. SSE(Server-Sent-Events)

SSE는 서버와 한번 연결을 맺고 나면, 일정 시간 동안 서버에서 변경이 발생할 때마다 서버에서 클라이언트로 데이터를 전송하는 방법입니다.

SSE(Server-Sent-Events)

과정은 다음과 같습니다.

• 클라이언트는 서버를 구독한다.(SSE Connection을 맺는다.)
• 서버는 변동사항이 생길 때마다 구독한 클라이언트들에게 데이터를 전송한다.

SSE는 상황에 따라서 응답마다 다시 요청을 해야 하는 Long Polling 방식보다 효율적입니다. SSE는 서버에서 클라이언트로 text message를 보내는 브라우저 기반 웹 애플리케이션 기술이며 HTTP의 persistent connections을 기반으로 하는 HTML5 표준 기술입니다. 그러나 HTTP를 통한 SSE(HTTP/2가 아닐 경우)는 브라우저 당 6개의 연결로 제한되므로, 사용자가 웹 사이트의 여러 탭을 열면 첫 6개의 탭 이후에는 SSE가 작동하지 않는다는 단점도 있긴 합니다. (HTTP/2에서는 100개까지의 접속을 허용합니다.)

전반적으로 SSE는 클라이언트가 서버와 크게 통신할 필요 없이 단지 업데이트된 데이터만 받아야 하는 실시간 데이터 스트림에 대한 구현이 필요할 때는 매우 훌륭한 선택입니다.
 

4. WebSockets

WebSockets은 OSI 네트워크 모델의 4 계층 프로토콜인 TCP에 기반한 양방향 메시지 전달 프로토콜입니다. WebSockets은 프로토콜 오버헤드가 적고, 네트워크 스택에서 더 낮은 수준에서 동작하기 때문에 HTTP보다 데이터 전송이 빠릅니다.

WebSockets

상위 수준에서 보면 WebSockets의 과정은 다음과 같습니다.

• 클라이언트와 서버는 HTTP를 통해 연결을 설정하고, WebSockets 핸드셰이크를 통해 연결한다.
• WebSockets의 데이터는 클라이언트 서버와 양방향으로 전송된다.

웹 소켓의 가장 큰 장점은 속도입니다. 클라이언트와 서버는 메시지를 전송할 때마다 서로의 연결을 찾아서 다시 설정할 필요가 없습니다. 웹 소켓 연결이 설정되면 데이터는 어느 방향으로든 즉시 안전하게 전송될 수 있습니다.(TCP이기에 메시지가 항상 순서대로 도착하도록 보장됨)

하지만 단점은 초기 구현에 상당히 많은 비용이 들어간다는 점입니다. 웹소켓은 멀티 온라인 게임과 같은 실시간 상태 업데이트와 긴밀한 동기화를 통해 분산된 사용자에게 전송해야 하는 경우 유용합니다.

따라서 전반적으로 웹 소켓은 빠른 고품질의 양방향 연결이 필요하다면 좋은 선택이지만, 시스템에 상당한 복잡성을 추가하고 구현하는 데 많은 투자가 필요하므로 폴링이나 SSE가 적합하지 않은 경우에만 사용하는 것을 권장합니다.
 

Spring에서 SSE로 알림 기능 구현하기

서론이 길었습니다. 그러면 이제 바로 Spring에서 SSE를 활용해 알림 기능을 구현하는 방법에 대해 작성해 보겠습니다.
전체 코드는 깃허브에서 확인 가능합니다.

NotificationController

@RestController
@RequestMapping("/notifications")
@RequiredArgsConstructor
public class NotificationController {
    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");
    }
}

먼저 NotificationController입니다. 간단하게 클라이언트에서 구독을 하기 위한 subcribe 메서드와 임시로 서버에서 클라이언트로 알림을 주기 위한 sendData 메서드를 생성했습니다.

NotificationService

@Service
@RequiredArgsConstructor
public class NotificationService {
    // 기본 타임아웃 설정
    private static final Long DEFAULT_TIMEOUT = 60L * 1000 * 60;

    private final EmitterRepository emitterRepository;

    /**
     * 클라이언트가 구독을 위해 호출하는 메서드.
     *
     * @param userId - 구독하는 클라이언트의 사용자 아이디.
     * @return SseEmitter - 서버에서 보낸 이벤트 Emitter
     */
    public SseEmitter subscribe(Long userId) {
        SseEmitter emitter = createEmitter(userId);

        sendToClient(userId, "EventStream Created. [userId=" + userId + "]");
        return emitter;
    }

    /**
     * 서버의 이벤트를 클라이언트에게 보내는 메서드
     * 다른 서비스 로직에서 이 메서드를 사용해 데이터를 Object event에 넣고 전송하면 된다.
     *
     * @param userId - 메세지를 전송할 사용자의 아이디.
     * @param event  - 전송할 이벤트 객체.
     */
    public void notify(Long userId, Object event) {
        sendToClient(userId, event);
    }

    /**
     * 클라이언트에게 데이터를 전송
     *
     * @param id   - 데이터를 받을 사용자의 아이디.
     * @param data - 전송할 데이터.
     */
    private void sendToClient(Long id, Object data) {
        SseEmitter emitter = emitterRepository.get(id);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event().id(String.valueOf(id)).name("sse").data(data));
            } catch (IOException exception) {
                emitterRepository.deleteById(id);
                emitter.completeWithError(exception);
            }
        }
    }

    /**
     * 사용자 아이디를 기반으로 이벤트 Emitter를 생성
     *
     * @param id - 사용자 아이디.
     * @return SseEmitter - 생성된 이벤트 Emitter.
     */
    private SseEmitter createEmitter(Long id) {
        SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
        emitterRepository.save(id, emitter);

        // Emitter가 완료될 때(모든 데이터가 성공적으로 전송된 상태) Emitter를 삭제한다.
        emitter.onCompletion(() -> emitterRepository.deleteById(id));
        // Emitter가 타임아웃 되었을 때(지정된 시간동안 어떠한 이벤트도 전송되지 않았을 때) Emitter를 삭제한다.
        emitter.onTimeout(() -> emitterRepository.deleteById(id));

        return emitter;
    }
}

다음은 NotificationService입니다.

클라이언트에서 처음 구독 시, 즉 subscribe 메서드에서 처음 구독 시에 sendToClient()를 통해 데이터를 전송하는 이유는 처음 SSE 응답을 할 때 아무런 이벤트도 보내지 않으면 재연결 요청을 보내거나, 연결 요청 자체에서 오류가 발생하기 때문입니다. 따라서 첫 SSE 응답을 보낼 시 더미 데이터를 넣어 이러한 오류를 방지하기 위해 전송합니다.

EmitterRepository

@Repository
@RequiredArgsConstructor
public class EmitterRepository {
    // 모든 Emitters를 저장하는 ConcurrentHashMap
    private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();

    /**
     * 주어진 아이디와 이미터를 저장
     *
     * @param id      - 사용자 아이디.
     * @param emitter - 이벤트 Emitter.
     */
    public void save(Long id, SseEmitter emitter) {
        emitters.put(id, emitter);
    }

    /**
     * 주어진 아이디의 Emitter를 제거
     *
     * @param id - 사용자 아이디.
     */
    public void deleteById(Long id) {
        emitters.remove(id);
    }

    /**
     * 주어진 아이디의 Emitter를 가져옴.
     *
     * @param id - 사용자 아이디.
     * @return SseEmitter - 이벤트 Emitter.
     */
    public SseEmitter get(Long id) {
        return emitters.get(id);
    }
}

다음은 EmitterRepository입니다. 모든 Emitter들을 저장하는 emitters객체를 thread-safe 한 ConturrentHashMap을 통해 선언했습니다.
 

테스트 

구독하기

먼저 GET "/subscribe/{id}"를 통해 구독을 시도하면 다음과 같이 출력됩니다.

데이터 발행하기

그리고 POST "/send-data/{id}"를 통해 서버에서 임시로 이벤트를 발행시킵니다.

발행된 데이터 전송왔는지 확인하기

그러면 구독한 클라이언트에서 이벤트를 전송받을 수 있습니다.

클라이언트를 기준으로 테스트를 해볼까요? 클라이언트는 먼저 아래와 같은 이벤트 구독을 위해 다음과 같은 코드를 사용할 수 있습니다.

클라이언트 코드

const eventSource = new EventSource('http://localhost:8080/notifications/subscribe/1');

eventSource.addEventListener('sse', event => {
    console.log(event);
});

WebConfig

@Configuration
public class WebConfig implements WebMvcConfigurer {
    public static final String ALLOWED_METHOD_NAMES = "GET,HEAD,POST,PUT,DELETE,TRACE,OPTIONS,PATCH";

    @Override
    public void addCorsMappings(final CorsRegistry registry) {
        registry.addMapping("/**")
                .allowedMethods(ALLOWED_METHOD_NAMES.split(","));
    }
}

cors 문제가 발생할 수도 있으니 서버에서 임시적으로 cors를 disable 처리했습니다.

이벤트 구독

그리고 console 부분에 js 코드를 그대로 삽입하면 이벤트 구독이 정상적으로 되는 것을 확인할 수 있습니다.

Response를 보면 Content-Type이 text/event-stream으로 된 것을 확인할 수 있습니다.

데이터 발행하기

그리고 마찬가지로 데이터를 발행시켜 보겠습니다.

이벤트 확인

이벤트를 성공적으로 수신한 것을 확인할 수 있습니다. 
 

결론

단순 알림 확인 API를 만들어 놓고 프론트에서 주기적으로 호출하도록 하는 Polling 방식이 구현에는 간단하긴 하지만, 여유가 된다면 SSE 방식으로 해보는 것도 좋은 경험이 될 것이라 생각합니다.

또한 위의 코드들은 정말 단순한 예제코드이기 때문에 실 서비스에서 운영할 것이라면 조금 더 아티클을 찾아보고 적용할 것을 권장드립니다. 다중 WAS 환경에서는 문제가 되는 코드이기 때문에 Redis의 Pub/Sub나 Kafka를 덧붙여서 사용하면 좋을 것 같습니다. 

추가로, OSIV가 켜져있을 시 SSE 서비스단에 트랜잭션이 걸려있다면 SSE연결 동안 트랜잭션을 계속 물고 있어 커넥션 낭비가 일어날 수 있으니 트랜잭션을 걸지 않아야 합니다.

 

참고

• https://tecoble.techcourse.co.kr/post/2022-10-11-server-sent-events/
• https://igotanoffer.com/blogs/tech/polling-sse-websockets-system-design-interview