[Spring Boot] SSE 를 통한 대기열 서비스 개선, 불필요한 폴링 제거

 

2024.09.01 - [Programming/Spring] - [Spring Boot] 트래픽 우회, 대기열 서비스 구현

[Spring Boot] 트래픽 우회, 대기열 서비스 구현

 

이전에 대기열 서비스에 대한 글을 발행했었다.

 

마지막 개선점을 다시 한번 확인해보면 G 마켓에서는 폴링 방식을 사용하고 있어 그대로 사용하였지만 기본적으로 실시간성 데이터에 허점이 존재하며, 서버의 부담도 크다는 것을 깨달았다. 

 

실시간성 데이터의 허점이라하면 

 

A 라는 사용자는 0.5초, 1.5초, 2.5초 ... 이런식으로 풀링하며,

B 라는 사용자는 0.3초, 1.3초, 2.3초 ... 이런식으로 폴링한다 가정하자.

 

만약 A 는 0.1 초에 대기열에 진입하였고, B 라는 사용자는 이보다 늦게 0.2 초에 진입했다고 하자. 만약 A, B 모두가 1초에 대기열을 벗어나 작업이 가능한 상태가 되었다면 누가 먼저 진입할 수 있을까?

 

원래대로라면 A 가 먼저 진입했기 때문에 B 보다 먼저 선착순 화면에 진입해야한다. 하지만 A 는 1.5초, B 는 1.2 초에 진입하면서 실제로는 B 가 더 빠르게 진입할 수 있다. 1초라는 단위시간 동안 화면에 진입하는 속도는 그렇게 큰 의미를 두지 않을 수 있지만, 선착순 이벤트에서는 보다 민감하게 받아드려야한다고 생각한다.

 

두 번째로 폴링 방식은 요청-응답이 1:1 을 이룬다. 그렇기 때문에 불필요하게 지속적인 요청이 계속해서 발생하게된다. 이 부분 또한 리소스 낭비라고 볼 수 있기에 개선의 여지가 있다. 

 

실시간성 보장 & 리소스 관리

우선 실시간을 보장하기 위해서는 흔히 WebSocket 과 SSE 방식이 존재한다. 두 방식 모두 폴링 방식보다 더 적은 리소스를 사용한다. 

 

WebSocket

웹소켓을 먼저 살펴보자면 HTTP 가 아닌 ws 이라는 새로운 프로토콜을 사용한다. 

처음 웹소켓 연결을 생성하면 HTTP 처럼 많은 헤더 등, 불필요한 요소를 없애고 데이터만을 주고받을 수 있다.

데이터를 주고받기에 양방향 데이터를 제공한다.

 

단점으로는 패킷 검사 기능이 있는 방화벽에서 WebSocket 을 처리하는데 문제가 있습니다.

 

Server Sent Events

SSE 는 서버에서 클라이언트로 단방향으로 실시간 이벤트를 전송하는 웹 기술이다.

말 그래도 단방향이며, 웹 기술이므로 HTTP 를 사용한다. 

처음 HTTP 요청이 들어오면, 서버는 HTTP 응답을 유지한 상태에서 데이터를 전송한다. 폴링 방식과 비교했을 때, 효율적이며 불필요한 통신을 최소화할 수 있다. 

 

WebSocket vs SSE ??

그럼 무엇을 사용해야하는 가??

 

웹소켓은 양방향 통신이 필요할 때, 송/수신 데이터가 많은 채팅 같은 서비스에 사용할 수 있다.

 

반면 SSE 는 불필요한 폴링을 줄이는 것이 우선적 목적이며, 개인화된 데이터를 이용한 이벤트 푸시를 할 때 사용할 수 있다.

 

그래서 대기열에서는??

대기열을 확인하는 과정은 우선 단방향으로 이루어진다.

 

또한 개인별 토큰을 사용해서 대기열 번호를 식별하고 있다. 만약 이를 웹소켓으로 구현한다면 토큰별 채널을 열어두고 해당 채널에 지속적으로 메세지를 발행해야할 것이다. 이보다는 SSE 에서 개인화된 데이터를 사용하는 것이 조금 더 적합해 보인다.

 

결국 현재 구현하는 서비스가 단순한 순번 알림 정도이므로 SSE 를 통해 단순하게 재구현하도록 하겠다.

 

Controller

@RestController
@RequestMapping("/queue")
public class QueueController {

    @Autowired
    private QueueService queueService;

    // SSE 연결 생성
    @GetMapping(value = "/sse-status/{token}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter getQueueStatus(@PathVariable String token) {
        SseEmitter emitter = queueService.addEmitter(token);
        return emitter;
    }
}

 

아주 단순한 형태로 구현해보겠다.

 

SSE 연결 요청이 들어오면 SseEmitter 를 생성하고 이를 등록한다. 

 

SseEmitter 란?

• Spring Framework 4.2 부터 SSE 프로토콜을 지원하기 위한 구현체 SseEmitter() 를 제공한다.
• 해당 구현체는 Content-Type, 데이터 포맷을 자동으로 맞춰주기에 편리하게 사용할 수 있다!

Scheduler

@Component
public class QueueScheduler{

    private final RedisTemplate<String, String> redisTemplate;
    private final ConcurrentHashMap<String, SseEmitter> sseEmitters;

    public QueueScheduler(RedisTemplate<String, String> redisTemplate, ConcurrentHashMap<String, SseEmitter> sseEmitters) {
        this.redisTemplate = redisTemplate;
        this.sseEmitters = sseEmitters;
    }

    @Scheduled(fixedRate = 1000) // 1초마다 실행
    public void updateQueueStatus() {
        Set<String> topTokens = redisTemplate.opsForZSet().range("WaitingQueue", 0, 999);
        int rank = 1;

        for (String token : topTokens) {
            SseEmitter emitter = sseEmitters.get(token);
            if (emitter != null) {
                try {
                    emitter.send("Token: " + token + " is at position " + rank);
                } catch (IOException e) {
                    emitter.completeWithError(e);
                    sseEmitters.remove(token);
                }
            }
            rank++;
        }
    }
}

 

실제 이전 글에서 보면 1초마다 N개의 토큰을 대기열에서 작업열로 이동시키는 작업을 진행했다. 

SSE 가 실시간성을 가지기 위해서는 토큰이 이동되면 즉시 1초마다 지정한 M 개의 토큰에 대해서 데이터를 전달해줘야한다. 
만약 작업열에 입장해도 되는 토큰이 응답을 받지 못했다면 M + a 개의 데이터를 1초마다 보내줘야한다. 

결국 대기열이라고 하는 서비스 자체가 트래픽을 처리하기 위한 서비스이지만 많은 리소스를 사용할 수 밖에 없는 서비스같다.
대기열 도입 시 이점을 확실히 파악하는 것이 필요할 것 같다.