배경

이전 글 참고

post를 통한 분석

SSE를 적용하지 않았던 모습입니다.

위의 사진으로 부터 아래 사진까지 10초 정도의 시간이 걸렸습니다.

문제는 단순 POST를 통해 분석을 요청하고 이 결과를 받으면 10초 동안 아무것도 하면 안 됩니다.

다른 페이지를 갔다 오지도 못하고 그 페이지에서 기다려야 결과를 받아볼 수 있습니다.

그래서 저번 포스트에서 작성한 SSE를 사용하도록 하겠습니다.

SSE 적용

// react
sse = new EventSource(`${process.env.REACT_APP_API_URL}/ai/emitter`); // 1
eventSource.addEventListener('data', (e) => { // 2
	console.log("이벤트 발생");
});

1. 서버에게 SSE 연결을 요청합니다.
2. data라는 이름으로 이벤트가 발생하면 수행할 액션을 정의합니다.

// spring
@GetMapping(value = "/api/ai/emitter", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter create(AccessTokenPayload ap) {
	return aiService.createEmitter(ap.getUid()); // ap.getUid() => 요청으로 부터 UID를 추출한다.
}


public SseEmitter createEmitter(UUID uid) {
        SseEmitter emitter = new SseEmitter();
        return emitterRepository.save(uid, emitter);
}

AccessTokenPayload는 제가 요청에서 유저의 정보를 추출할 수 있도록 Argument Resolver를 사용한 것 입니다. 없다고 생각하셔도 됩니다.

다만 SSE에 연결하기 위해선 GET을 지원하며 POST는 지원하지 않기에 SSE 연결에 필요한 추가 데이터를 보내주고 싶다면 Path String을 사용하는 등 추가적인 데이터를 보낼 수 있는 수단을 생각하셔야 합니다.

+

추가로 왜 User Id(UID)가 필요한지 말씀드리겠습니다.

AI 분석을 한 클라이언트만 요청하는 것이 아닙니다. 결국 많은 곳에서 요청을 보내고 이에 따라 SSE 연결이 많아집니다.

Broadcast가 아니라 단 한명에게만 이벤트를 보내야합니다.

특정 유저에게 알림 이벤트를 보내야 하기에 User Id에 따른 이벤트를 발생시킬 Emitter가 다르게 됩니다.

그래서 User Id와 Emitter를 함께 저장해서 어떤 유저의 분석이 끝났는지 확인하고 이벤트를 발생시킬 수 있도록 합니다.

@Repository
public class EmitterRepository {
    static Map<UUID, SseEmitter> emitterMap = new ConcurrentHashMap<>();
    public SseEmitter save(UUID uid, SseEmitter emitter) {
        emitterMap.put(uid, emitter);
        return emitter;
    }

    public void deleteById(UUID uuid) {
        emitterMap.remove(uuid);
    }

    public SseEmitter getEmitterByUserId(UUID uid) {
        return emitterMap.get(uid);
    }
}

Emitter를 저장할 곳을 생각해봤습니다. mysql에 넣자니 길어봐야 10초 남짓한 연결을 위해 mysql에 저장하는 것은 너무 소비가 크다고 생각했습니다.

애초에 mysql에는 emitter를 저장할 타입도 없구요.

똑같이 redis에 저장할 수 없었던 이유도 이를 저장할 타입을 지원하지 않기 때문이었습니다.

그래서 어차피 짧은 시간동안 연결을 유지하기에 단순 Map을 사용하는 것을 생각했습니다.

애초에 프로그램이 꺼져 Emitter 데이터가 휘발되어도 다시 연결하면 그만이기도 하고요.

+

ConcurrentHashMap이란 Java 1.5 버전에서 HashTable의 대안으로 처음 소개된 Collection입니다.

Map 전체에 synchronized를 걸어 Thread-safe하지만 성능에 저하가 있던 기존의 HashTable 또는 synchronized map과 달리 ConcurrentHashMap은 쓰기 작업에만 synchronized를 걸고 get에는 걸지 않아 읽기에 여러 쓰레드에서 동시 접근이 가능해 성능이 더 개선되었습니다.

문제 1

하지만 결국 실패했습니다.

SSE 연결을 요청했지만

연결이 되지 않고 실패합니다.

해결 1

이러한 증상은 많은 사람들이 공통적으로 겪는 문제였습니다.

해결방법은 간단합니다.

SSE 연결하고 Timeout이 발생하기 전에 데이터를 보내야 문제가 발생하지 않는다고 합니다.

즉 SSE를 연결하고 더미 이벤트를 발생시켜 데이터를 보내야 정상 연결이 수행됩니다.

public SseEmitter createEmitter(UUID uid) {
    SseEmitter emitter = new SseEmitter();
    try {
    	emitter.send(SseEmitter.event()
        	.id(String.valueOf(uid))
            .data("Dummy data"));
	} catch (IOException e) {
    	emitterRepository.delete(uid);
        emitter.completeWithError(e);
	}
    return emitterRepository.save(uid, emitter);
}

SSE 연결을 하고 Dummy Data를 보내면 문제는 해결됩니다.

문제 2

위와 같이 진행하면 SSE 연결과 이벤트 발생 자체는 문제없이 됩니다.

다만 한가지 문제를 발견했습니다.

이벤트가 발생하면 즉 분석이 완료됐다는 이벤트를 수신했다면 더 이상 SSE 연결을 할 필요가 없습니다. 지워버려야 하는 것이 정상입니다.

다만 이벤트를 수신했는데도 SSE 연결 요청을 지속적으로 보내고 있는 것을 확인했습니다.

해결 2

원인은 Timeout입니다.

설정한 timeout이 지나면 클라이언트에서는 SSE 재연결을 요청합니다. 왜냐하면 SSE는 연결을 끊지 않는 한 계속 지속되어야 하니까요.

SSE 연결을 위해 객체를 생성할 때 아무런 인자를 넘기지 않았습니다.

SseEmitter emitter = new SseEmitter(); // 아무런 인자가 없음

이러한 상황에서 재연결을 위에서 보시다 싶이 30초에 한번씩 보내는 것을 확인할 수 있습니다. 즉 기본적으로 30초의 시간을 갖는다는 것 입니다.

그래서 인자로 1분을 넘겨주면 1분에 한 번씩 연결 재요청을 보내는 것을 확인할 수 있습니다.

SseEmitter emitter = new SseEmitter(60000L);

다시 돌아와서 설명하자면 문제는 이벤트를 수신했지만 연결이 지속되어 Timeout 지났고, 이에 따라 재연결 요청을 하는 것 입니다.

즉 연결을 끊어주는 부분이 필요하다는 것이죠.

# react
sse = new EventSource(`${process.env.REACT_APP_API_URL}/ai/emitter`); // 1
eventSource.addEventListener('data', (e) => { // 2
	console.log("이벤트 발생");
    sse.close();
});

react 코드에서 이벤트 수신 시 sse.close()를 명시해줌으로써 SSE 연결을 해제할 수 있고 시간이 지나도 재연결 요청을 보내지 않습니다.

+

Timeout을 너무 작게 설정하면 너무 과한 요청을 지속적으로 보낼 것 입니다. 즉 적당하게 잡아주는 것이 필요한데,

위에서 말씀드렸다싶이 AI 분석은 한번에 10초정도를 사용합니다. 그래서 저는 넉넉하게 2배인 20초를 설정해줬습니다.

Timeout 설정은 서비스에 따라 다르니 적절한 시간을 설정하는 것이 좋겠습니다.

다음편에 계속