알림 기능 Polling 에서 SSE 전환하며 겪었던 삽질 (사이드 프로젝트)

기존 MVP 모델에서 실시간 알림을 위해 폴링 기반으로 메시지 알림을 구현했다.

 

20초 주기로 폴링(polling) 설정하여 알림 메시지 갯수를 전달하는 방식이었고,

AWS 프리티어 기반 최소 스펙으로 서버를 구성하고 있어 비용과 부하를 최소화하기 접근 방식이었다.

 

하지만 알림 기능의 최소한의 실시간성을 보장하지 못하는 피드백이 있어 다른 접근 방법이 필요했다.

 

최소한의 실시간성을 보장하기 위해 짧은 폴링 주기도 검토했지만,

낮은 스펙의 서버 환경에서 부하를 다시 증가하는 문제가 있어 해당 방법은 적합하지 않아 다른 접근 방식이 필요했다.

 

그래서 대안 중 하나로 SSE(server-sent-event) 를 검토한 과정과 이를 적용하는 과정에 적어보고자 한다.

압도적으로 많은 읽지 않는 알림 수 조회 API...

 

 

1. 실시간성을 위한 선택: WebSocket vs SSE ??

최소한의 실시간성을 보장하기 위해 WebSocket 과 SSE 을 검토하기 이전에 비즈니스 특성을 검토해보기로 했다.

 

현재 알림 메시지 특성은 개인화된 데이터를 기반한 이벤트 푸시가 많았다.

예를 들어, 게시글에 리뷰 생성 또는 체크리스트의 상태가 변경되면

작성자에게 알림이 전달되는 형태 모두 개인화된 알림을 전달하는 방식이었다.

 

두 방식 모두 불필요한 폴링을 최소화할 수 있다.

 

WebSocket 의 경우 양방향성 통신에 관한 요구사항을 충족하기 위한 일반적인 접근 방법이다.

하지만, 현재 비즈니스 로직 특성을 고려하면 서버에서 사용자에게 단방향 메시지를 전달하는 방식이므로

서버에서 클라이언트로의 단뱡향 통신에 관한 요구사항만을 충족하는 SSE 사용이 적합하다고 판단했다.

 

추가적으로 SSE는 추가적인 프로토콜에 관한 지식이 필요하지 않아 기술 적용의 허들이 낮은 장점이 있었다.

기존 HTTP protocol 을 기반한 TCP Connection 유지하는 방식으로 동작하기 때문이었다.

(참고로 Websocket 은 ws protocol 기반으로 동작하므로 추가적인 지식이 필요하다.)

 

(참고)

SSE 는 Auto Reconnect 지원과 Cookie 첨부 요청이 가능하고 응답 형태를 text 로 전달하는 특성을 가진다.

반면, WebSocket 의 경우에는 Auto Reconnect 지원을 하지 않고 text 또는 binary 형태로 데이터 전달이 가능하다.

 

 

2. SSE 동작 방식

SSE 또한 HTTP/1.1 기반으로 동작하기 때문에 TCP 기반의 연결 작업을 진행한다.  

1. TCP 3-way handshake 과정을 통해 TCP Connection 을 수립한다.
2. 클라이언트는 요청으로 HTTP 헤더에 Accept: text/event-stream 로 설정하여 요청한다.
3. 서버는 응답으로 HTTP 헤더에 Content-Type: text/event-stream 으로 전달한다.
4. 서버에서 클라이언트 쪽으로 지속적으로 스트리밍을 통해 데이터를 푸시한다.

 

 

3. SSE 주의 사항 - 연결 수립된 서버에게만 데이터 전달이 가능하다.

SSE 를 사용할 경우 메시지는 Connection 을 수립한 서버에서만 메시지 스트리밍이 가능하다.

 

TCP 는 연결지향 프로토콜이다.

TCP 연결은 Client  IP, Client Port, Server IP, Server Port 정보를 토대로 고유하게 식별된다.

4가지 정보를 통해 하나의 커넥션 소켓을 정의한다.

 

커널 메모리의 TCP 커넥션 상태 테이블(connection-state table) 에 해당 정보를 저장한다.

커널 메모리는 OS 단위로 관리되기 때문에 커넥션을 수립한 서버에서만 유지되며 다른 서버에서 이 연결 상태를 알 수 없다. (e.g. 버퍼, 순서, 패킷 번호 등)

State      Recv-Q  Send-Q      Local Address:Port         Peer Address:Port
------------------------------------------------------------------------
LISTEN     0       511         *:80                       *:*
SYN-RECV   0       0           192.0.2.145:80             203.0.113.5:35449
SYN-RECV   0       0           192.0.2.145:80             203.0.113.27:53599
ESTAB      0       0           192.0.2.145:80             203.0.113.27:33605
TIME-WAIT  0       0           192.0.2.145:80             203.0.113.47:50685

(1) State : TCP 상태 (LISTEN, SYN-RECV, ESTAB, 등)
(2) Recv-Q : 수신 큐에 쌓인 데이터 (읽히지 않음)
(3) Send-Q : 송신 큐에 쌓인 데이터 (아직 전송 안 됨)
(4) Local Address:Port : 서버 측 IP와 포트
(5) Peer Address:Port : 클라이언트 측 IP와 포트

 

 

4. Scale-out 환경에서 SSE 는 pub/sub 구조 도입을 고려해야 한다.

ALB 을 통해 로드 밸런싱하고 있다면  메시지가 SSE Connection 이 수립된 서버로 전달을 되지 않을 가능성이 높다.

 

ALB는 새 HTTP 요청이 도착할 때마다 L7 기반 로드 밸런싱 알고리즘에 따라 라우팅 할 서버를 선택한다.

(L7 기반 로드 밸런싱 알고리즘 : 라운드로빈, IP 해시, 가중치 등)

 

ALB 는 매 요청마다 새로 백엔드 서버를 선택하고 새 소켓을 생성하거나 기존 소켓 풀에서 해당 서버로 연결된 소켓을 선택해 사용한다.

즉, 소켓 테이블의 항목이 변경되는 게 아니라, ALB 가 관리하는 커넥션 풀 내에서 다른 서버와 연결된 소켓을 선택해 사용하는 것이다.

 

이와 같은 이유로 인해 스케일 아웃 환경에서 커넥션 수립된 서버에서 메시지 스트리밍을 지원 가능하게 하기 위한 고민이 필요하다.

이와 같이 클라이언트 정보를 가진 서버가 메시지 적절하게 메시지 푸시하기 위해 pub/sub 구조를 통해 해당 서버가 스트리밍하기 위한 메시지를 구독하고 해당 메시지가 스트리밍 해야하는 메시지를 판별하여 푸시하도록 구성할 수 있다.

스케일 아웃 환경에서 SSE 기반 메시지 푸시 지원 방법 : pub/sub

 

 

5. Spring 에서 SSE 도입 시 동시성을 고려해야 하는 이유

스프링에서는 SseEmitter 를 제공해 손쉽게 SSE 를 사용할 수 있다. 해당 SseEmitter 를 생성 후, 메시지를 전달하는 기능을 제공한다.

사용자별 또는 기기별로 Emitter 를 관리하기 위해서는 애플리케이션 메모리에 Emitter 를 관리해야 한다.

final SseEmitter sseEmitter = new SseEmitter(600 * 1000L);

 

한가지 주의해야 할 점은 동시성을 고려해야 한다.

 

SseEmitter 에서 제공하는 콜백 함수인 onCompletion, onTimeout, onError 메서드는 별도의 스레드에서 비동기적으로 동작한다

 

그러므로 현재 요청 중을 처리 중인 스레드와 동시성 이슈가 발생할 가능성이 있으므로 동시성을 고려한 설계가 필요하다.

대표적인 예시로는 Emitter 를 관리하는 자료 구조를 사용할 경우 동시성 제어를 지원하는 자료구조 사용을 권장한다.

// key : memberId
private final Map<UUID, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

 

아래 SseEmitter 를 구성했던 예시 코드이다.

@Slf4j
@Component
public class SseEmitters {

	// key : memberId
	private final Map<UUID, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

	public SseEmitter add(UUID clientId) {
		final SseEmitter sseEmitter = new SseEmitter(600 * 1000L);

		if (sseEmitters.containsKey(clientId)) {
			return sseEmitters.get(clientId);
		} else {
			sseEmitters.put(clientId, sseEmitter);
		}

		sseEmitter.onCompletion(() -> {
			log.info("onCompletion sseEmitter : {}", sseEmitter);
			sseEmitters.remove(clientId);
		});
		sseEmitter.onTimeout(() -> {
			log.info("onTimeout sseEmitter : {}", sseEmitter);
			sseEmitters.remove(clientId);
		});
		sseEmitter.onError(e -> {
			log.info("onError sseEmitter : {}, error: {}", sseEmitter, e.getMessage());
			this.sseEmitters.remove(clientId);
		});

		log.info("added sseEmitter id : {}, sseEmitters size: {}", clientId, sseEmitters.size());
		return sseEmitter;
	}

	public SseEmitter remove(UUID memberId) {
		if (!sseEmitters.containsKey(memberId)) {
			throw new NotificationEmitterNotFoundException(NotificationErrorType.EMITTER_NOT_FOUND);
		}

		log.debug("removing sseEmitter for memberId: {}", memberId);

		final SseEmitter removedEmitter = sseEmitters.remove(memberId);
		removedEmitter.complete();

		return removedEmitter;
	}

	public <T> boolean send(UUID clientId, String eventName, T data) {
		if (!sseEmitters.containsKey(clientId)) {
			return false;
		}

		SseEmitter emitter = sseEmitters.get(clientId);

		try {
			emitter.send( // 메시지 전달
				SseEmitter.event()
					.name(eventName)
					.data(data)
					.build());

			return true;
		} catch (Exception exception) {
			log.info("send sseEmitter error : {}", exception.getMessage());
			sseEmitters.remove(clientId);
			emitter.completeWithError(exception);
		}

		return false;
	}
}

 

6. 클라이언트는 일정 시간 메시지를 전달받지 못하면 연결이 끊어진다. (health check )

SSE 는 연결을 유지하여 메시지를 단방향으로 송수신하는 구조이다.

SSE 는 효율적인 연결을 위해 서버에서 클라이언트 방향으로 health check 를 해야 한다. 

그렇기 때문에 일정 시간동안 메시지를 클라이언트에서 수신받지 못하면 연결을 해제하기 때문에 health check 를 위한 임의의 메시지를 주기적으로 전달하는 로직을 구성해야 한다. (default : 45초)

 

 

Reference

• [naver d2] TCP/IP 네트워크 스택 이해하기 : https://d2.naver.com/helloworld/47667
• [TOSS] SLASH 24 - SSE 이벤트 푸쉬로 불필요한 Polling 제거하기 : https://www.youtube.com/watch?v=ovGgdPPUZ2I
• TCP Connection States : https://maxnilz.com/docs/004-network/003-tcp-connection-state/
• [tecoble] Spring에서 Server-Sent-Events 구현하기 : https://tecoble.techcourse.co.kr/post/2022-10-11-server-sent-events/