실시간 메시지 전송의 오버헤드
구현하고 있는 이커머스 플랫폼에서는 SSE를 사용해서 사용자에게 알림을 보내는 기능을 가지고 있습니다. 알림 내역을 DB에 저장하고, 이후에 성공적으로 알림이 전송되면 SSE를 전송하는 구조를 취하고 있는데, 여기서 알림 내역을 저장한는 부분에 대한 테스트를 어느정도 진행이 되었기 때문에 잠시 접어두고 연결된 Emitter에 메시지를 전송하는, 실시간성 기능에 대한 스트레스 테스트를 진행해보려고 합니다. 이전에 로컬호스트 테스트에서 한 번에 접속 가능한 Emitter 개수는 3만개였고, 이를 개선하고자 여러가지 시도를 한 바 있습니다.
https://soberyl.tistory.com/37
[Test] 점진적 부하테스트: 30,000명 유저의 SSE 구독 요청 - TCP
이전 포스트에 이어서 SSE 구독 요청 시나리오를 계속해서 진행합니다.https://soberyl.tistory.com/36 [Test] 점진적 부하테스트: 10,000명 유저의 SSE 구독 요청 - 톰캣저번 포스트에 이어서, SSE 구독 요청을
soberyl.tistory.com
본 포스트에서는 SSE 구독에 대한 오버헤드 자체는 최대한 배제하고, Send하는 오버헤드 그 자체에 대해서 다루려고 합니다. 단위별로 쪼개서 부하를 관찰하고, 통합 테스트에서 기대하는 값이 나오는지 측정해보겠습니다. 기존의 코드를 살펴보면 다음과 같습니다.
@Transactional(readOnly = false)
public void sendToAllMembers(Long templateId, Map<String, String> variables) {
futures.add(CompletableFuture.runAsync(() -> {
LocalDateTime now = LocalDateTime.now().minusSeconds(1);
memberNotificationJdbcRepository.partitionInsert(templateId, variables, currentStart, currentEnd); // DB 작성
List<NetworkMessageDto> chunkData =
memberNotificationJdbcRepository.findNetworkMessageDtoByRange(templateId, currentStart, currentEnd, now); // 실시간 메시지 객체 생성
if (!chunkData.isEmpty()) {
eventPublisher.publishEvent(new NotificationSavedEvent(chunkData)); // 실시간 메시지 전송 이벤트 발송
}
}, executorService));
}
DB 작성이 완료되면 비동기 이벤트를 호출하게 됩니다. 요청이 많아지면 chunk 단위로 잘라서 멀티 스레드로 send를 진행하게 될 것입니다. 호출한 비동기 이벤트는 servlet의 sse emitter의 send를 호출하게됩니다. 해당 부분의 코드를 SseEmitter.java에서 살펴보면 아래와 같습니다.
// SseEmitter.java
public void send(SseEventBuilder builder) throws IOException {
Set<DataWithMediaType> dataToSend = builder.build();
this.writeLock.lock();
try {
super.send(dataToSend);
}
finally {
this.writeLock.unlock();
}
}
// ResponseBodyEmitter.java
protected final Lock writeLock = new ReentrantLock();
// org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter
private <T> void sendInternal(T data, @Nullable MediaType mediaType) throws IOException {
if (data instanceof ModelAndView mav) {
this.fragmentHandler.handle(mav);
return;
}
for (HttpMessageConverter<?> converter : this.messageConverters) {
if (converter.canWrite(data.getClass(), mediaType)) {
((HttpMessageConverter<T>) converter).write(data, mediaType, this.outputMessage);
return;
}
}
throw new IllegalArgumentException("No suitable converter for " + data.getClass());
}
dataToSend라는 DataWithMediaType 객체를 생성하고, write lock을 걸고 send를 하고 unlock합니다. 여러 스레드가 동시에 하나의 Emitter에 .send()를 호출할 경우에 메시지가 섞이거나 (Interleaving) 깨지는 것을 방지하기 위해서 한 번에 하나의 메시지만 쓰도록 동기화를 보장하는 것입니다.
sendInternel은 커널 버퍼에 패킷 내용을 적재할 때 호출되는 메서드인데, converter.write(data, mediaType, this.outputMessage); 여기서 객체를 직렬화합니다. Spring Boot를 사용 중이라면 여기서 Jackson을 이용해서 MappingJackson2HttpMessageConverter을 이용해서 변환하게 됩니다.
synchronized(this){ ... }를 사용하게 되면 가상 스레드를 지원하지 않습니다. Java 버전이 높아질 수록 이 ReentrantLock으로 코드를 대체하고 있습니다. 다만, 모든 코드가 대체되지는 않을 것입니다. Sse emitter에서 사용되는 write에 대한 Lock은 ReentrantLock으로 사용되고 있습니다. 따라서 가상 스레드를 사용할 수 있습니다.
실시간 SSE 메시지 전송 시 예상되는 병목 지점
- 직렬화 오버헤드가 있을 수 있을 것 같습니다. 1만 명에게 동일한 알림을 보낸다고 한다면 emitter.send(object)를 호출할 때마다 내부적으로 Jackson 라이브러리가 작동하여 Java 객체를 JSON 문자열 변환을 수행하기 때문에, 동접자가 많을 수록 이 변환을 많이 반복할 것입니다. 이에 대한 CPU 사용이 폭증할 수 있습니다.
- Slow client로 인한 스레드 블로킹으로 인한 I/O 병목이 발생할 수 있습니다. SseEmitter는 기본적으로 블로킹 I/O 모델 기반입니다. send()는 데이터를 커널의 TCP 버퍼에 밀어넣는 작업인데, 만약 클라이언트가 데이터를 늦게 읽어가서 커널 버퍼가 가득 차면 서버의 스레드는 그 데이터가 비워질 때 까지 write 동작에서 대기하게 됩니다. 이때 waiting 상태에 빠져서 병목 현상이 발생할 수 있습니다.
- 동접자가 매우 많을 경우에 트래픽이 폭주하면 서버의 NIC가 처리할 수 있는 대역폭을 초과하게 되고, 이때 패킷 드랍이나 재전송이 발생합니다. 이에 전송 지연이 기하 급수적으로 늘 수 있습니다.
테스트 환경
K6 + Grafana + InfluxDB + Prometheus 로 구성해서 진행합니다.
실시간 SSE 메시지 전송 스트레스 테스트
export const options = {
scenarios: {
// (Subscribers)
audience: {
executor: 'ramping-vus',
exec: 'audience',
startVUs: 0,
stages: [
{ duration: '2m', target: n }, // 2분 동안 n 명 연결
{ duration: '5m', target: n }, // 5분 동안 연결 유지
],
gracefulStop: '30s',
},
director: {
executor: 'constant-arrival-rate',
exec: 'director',
rate: 1,
timeUnit: '15s',
duration: '5m',
preAllocatedVUs: 1,
startTime: '2m30s',
},
},
};
k6/x/sse를 사용해서 sse에 메시지에 대한 응답속도를 측정합니다. 원하는 만큼의 유저가 2분 동안 n명이 sse 연결이 된 이후에 5분 동안 연결을 유지하고, 이 와중에 director가 sse send 로직이 포함된 api를 요청합니다. 여기서 startTime은 k6가 시작된 시점부터 2분 30초 뒤 부터 director를 활성화하겠다는 옵션입니다. 이 시나리오에 의하면 전체 시나리오는 7분 30초가 걸리게 될 것입니다. timeUnit은 알림 전송 로직의 시간 간격이고, rate은 목표 횟수입니다. 즉 15초 간격으로 알림 요청을 1번씩 요청하게 될 것입니다.
sse.open(SSE_URL, params, (client) => {
...
client.on('event', (event) => {
if (!event.data || event.data.trim() === '') {
return;
}
const receivedAt = Date.now();
try{
const payload = JSON.parse(event.data);
if (payload.timestamp) {
const sentAt = payload.timestamp;
msgLatency.add(receivedAt - sentAt); // 지연 시간
}
msgCount.add(1);
} catch(e){
console.log(e);
}
});
...
};
Client 측에서 sse를 처리하게 될때 사용되는 이벤트 리스너를 k6에서도 적용할 수 있도록 k6/x/sse가 제공합니다. 여기서 event를 수신했을 때, payload에서 서버에서 send할 때 찍은 timestamp와 클라이언트에서 받았을 때 찍은 timestamp를 비교해서 지연 시간을 측정합니다. 동시에 msgCount도 측정을 해서, 모든 메시지에서 connect나 heartbeat와 같은 더미 메시지는 제거한 순수 이벤트 메시지가 얼마나 도착했는지 기록합니다.
시나리오 0: 50명의 유저에게 동시에 SSE 메시지를 전송한다.
테스트 결과
50명의 유저에게 테스트 삼아서 send 시나리오를 돌려봤습니다. 결과를 분석해보면 다음과 같습니다.
| 필드 | 수치 | 의미 |
| check_total | 20 | director trigger 요청 횟수 |
| check_succeeded | 100% | 요청에서 성공의 비율 |
| check_failed | 0% | 요청에서 실패의 비율 |
| sse_event | 1550 | 수신한 모든 sse event의 수 |
| sse_msg_count | 1450 | 실질적인 event 수 (heart beat 등 제외) |
| sse_msg_latency | 11ms | 클라이언트가 event를 수신하기 까지 걸린 시간 |
| http_req_duration | 53ms | director의 알림 trigger를 완료하는데 걸린 시간 |
목표는 제이콥 닐슨의 인지 임계값을 참고하여 설정하려고 합니다.
| 시간 | 느낌 | 설명 |
| ~ 0.1초 (100ms) | 즉각적 (Instant) | 내가 버튼을 누르자마자 불이 켜진 느낌. 인과관계가 완벽히 연결됨. |
| ~ 0.3초 (300ms) | 매끄러움 (Flow) | 아주 약간의 틈은 있지만, 생각의 흐름이 끊기지 않음. 대화나 채팅의 "빠릿한" 느낌. |
| ~ 1.0초 (1000ms) | 지연 감지 (Noticeable) | "어? 뭔가 처리하고 있네?"라고 느낌. 하지만 여전히 실시간 범주에는 포함됨. |
| 10초 이상 | 이탈 (Lost) | 집중력이 깨지고 딴생각을 하게 됨. 로딩 바가 필수. |
제가 진행하고 있는 부분은 실시간 알림인데, 정말 즉각적인 반응은 필요없습니다. 따라서 300ms로 threshold를 설정하고 진행하겠습니다. 최악의 경우에는 500ms의 딜레이를 받도록 하려고 합니다.
시나리오 1: 1,000명의 유저에게 동시에 SSE 메시지를 전송한다.
최초의 서버 상태
테스트 결과
50명에서 1000명으로 늘려서 테스트해보니, sse_msg_latency가 약 23배 늘었습니다. 230ms가 되었고, 최악의 상황에는 2.2초가 걸렸습니다. 피크 타이밍에 과한 시간이 걸렸습니다. 이는 Queueing delay가 누적됐거나, GC가 터져서 느려졌거나, 네트워크 버퍼가 차서 블로킹 된 스레드가 발생했거나 하는 시간인 것 같습니다. msg_latency가 236ms인 것은 괜찮은 사실이나, max가 2224ms라는 것은 문제가 있습니다. 상정한 스레시 홀드는 최대 500ms이기 때문에 개선이 필요할 것 같습니다.
16:32에 fd가 1200개가 되면서 1000명의 구독을 수용했고, 이때 send를 진행합니다. 여기서 CPU 연산이 증가했습니다. JVM에서 사용하는 메모리가 중요한데, 0.2 정도의 CPU를 사용하고 있습니다.
JVM 메모리도 테스트를 진행하면서 send를 하는 시점에 Eden이 계속해서 꽉차고 있습니다. minor gc가 5~6번 일어난 것으로 보입니다. 이때 Old로 이동한 메모리가 이동했는데, 이는 SSE emitter 객체가 이동한 것입니다. 또한 시간이 조금 지난 후에 관찰해보니, 그래도 2.9GB 정도가 Eden에 있었습니다. SseEmitter 객체는 1개당 약 1KB 정도로, 1000명이면 1MB입니다. 게다가 테스트 진행 중에 Old로 옮겨갔을 것이므로 Eden에 남아있는 것은 JSON 객체, HTTP 응답 객체, Tomcat의 객체일 것입니다. Eden이 꽉 찰 경우에 minor GC가 일어날 경우에는 10ms 정도 걸리기 때문에 문제는 크게 없습니다.
CPU가 튀어오르는 것을 보았을 때, CPU 계산을 어느정도 진행하는 것 같습니다. 작성한 send의 코드는 다음과 같습니다.
@Override
public void send(List<NetworkMessageDto> messages) {
for (NetworkMessageDto message : messages) {
Long receiverId = message.getReceiverId();
Map<String, SseEmitter> userEmitters = emitterRepository.findAllStartWithById(receiverId);
userEmitters.forEach((emitterId, emitter) -> {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
});
}
}
public NotificationSendResult sendBySse(SseEmitter emitter, String emitterId, String eventId, String eventName, Object data) {
try {
emitter.send(SseEmitter.event()
.id(eventId)
.name(eventName)
.data(data));
return NotificationSendResult.success(emitterId, eventId);
} catch (IOException | IllegalStateException e) {
emitterRepository.deleteById(emitterId);
return NotificationSendResult.failure(emitterId, eventId, e.getMessage());
}
}
이렇게 sendBySse에서 data를 객체로 받고 있습니다. 객체로 넘긴 data는 Jackson 라이브러리를 사용해서 직렬화해서 보내게 됩니다.
위의 코드에서 send 로직을 확인하면 NetworkMessageDto를 하나씩 참조해서 대상 Emitter에게 보내는데, N개의 클라이언트에게 보낼때 마다 같은 내용이더라도 직렬화를 반드시 N번 해야하는 로직 상의 단점이 있습니다.
...
Map<String, Object> content = new HashMap<>(
MessageUtil.createMessageByHashMap(COMMON_RECEIVE_NOTIFICATION_SUCCESS.getMessageTemplate())
);
content.put("timestamp", System.currentTimeMillis());
return jdbcTemplate.query(sql, (rs, rowNum) -> NetworkMessageDto.builder()
.receiverId(rs.getLong("member_id"))
.eventId(String.valueOf(rs.getLong("id")))
.eventName("notification")
.data(content)
.build(),
templateId, minMemberId, maxMemberId, batchStartTime
);
}
여기서 NetworkMessageDto를 builder로 생성하는데, 여기서 직렬화를 해서 content를 진행해서 보내준다면 생성된 스레드 횟수만큼만 직렬화를 하게 될 것입니다. 또한 루프 안에서 직렬화를 하게 되면 매번 직렬화에 사용되는 수많은 임시 객체가 생성되었다가 버려지므로 이를 줄인다면 Eden memory의 효율화도 가져갈 수 있을 것입니다. 특히 맨 마지막에 sse 이벤트를 수신하게되는 클라이언트의 응답 지연 속도가 많이 개선될 것으로 생각됩니다.
가설 (1)
데이터 전송 간 직렬화 횟수를 N번에서 (스레드 개수)번 만큼으로 줄인다면 CPU 계산 양이 줄면서 응답 지연 속도가 줄어들 것입니다.
가설 (1) 개선 시도
public List<NetworkMessageDto> findNetworkMessageDtoByRange(Long templateId, long minMemberId, long maxMemberId, LocalDateTime batchStartTime) {
String sql =
"SELECT n.member_id, n.id " +
"FROM member_notification n " +
"WHERE n.notification_id = ? AND n.member_id BETWEEN ? AND ? " +
"AND n.created_at >= ?";
System.out.printf("[Range: TemplateId=%d, MemberId=%d-%d]%n in %s actual time %s", templateId, minMemberId, maxMemberId, batchStartTime, now());
Map<String, Object> content = new HashMap<>(
MessageUtil.createMessageByHashMap(COMMON_RECEIVE_NOTIFICATION_SUCCESS.getMessageTemplate())
);
content.put("timestamp", System.currentTimeMillis());
String jsonPayload = objectMapper.writeValueAsString(content); // 스레드 별로 미리 압축
return jdbcTemplate.query(sql, (rs, rowNum) -> NetworkMessageDto.builder()
.receiverId(rs.getLong("member_id"))
.eventId(String.valueOf(rs.getLong("id")))
.eventName("notification")
.data(jsonPayload)
.build(),
templateId, minMemberId, maxMemberId, batchStartTime
);
}
가설 (1) 개선 결과
254ms로 시간이 더 결렸습니다... 고점이 2547ms인데, 전체적으로 20ms 더 늘어난 것으로 보입니다.
덜 사용할 것 같았던 메모리도 더 사용하고 있습니다.
그런데 오히려 CPU 사용량은 줄었습니다. 원래는 0.2 정도였는데, 지금은 0.03 정도에 그쳤습니다.
메모리 사용량도 늘었습니다. Old가 비슷한 것을 보면, 연산할 때 생겨나는 더미 객체들이 늘어난 것으로 보입니다.
* 왜 이런 일이 생겼는지는 Jackson의 Serialization 알고리즘에 대해서 알 필요가 있습니다.
https://soberyl.tistory.com/41
[Util] Jackson의 특징과 직/역직렬화 과정
What is Jackson? Jackson은 자바 객체와 JSON 사이를 통역해주는 JSON 라이브러리입니다. (JSON Java class) Jackson은 단순히 JSON을 직렬화, 역직렬화 등의 변환을 위한 것 뿐만 아니라 어떤 데이터 포맷이든 자
soberyl.tistory.com
직렬화가 필요한지 필요하지 않는지는 해당 Entity의 Anotation으로 판단합니다. 따라서 단순히 ObjectMapper로 먼저 직렬화를 해놓는다고 해도 Anotation을 붙여놓지 않으면 Jackson은 해당 필드를 위해서 Initialization부터 메모리를 할당하고 순회를 진행하면서 직렬화를 진행할 것입니다. 이래서 메모리 할당량이 증가한 것으로 보입니다. CPU 계산이 줄어든 것은, Data는 자바 객체였지만 이미 문자열로 만들어 놓았기 때문에 추가적인 Jackson의 BeanSerializer 단계가 생기지 않기 때문입니다. 대신 String 필드에 대해서 수행하는 직렬화 기능인 특수문자 유무를 확인하기 위한 문자열 전체 Scan을 진행하고 특수문자에 \"를 붙입니다. 거의 소요되는 시간은 비슷할 것입니다.
따라서 메모리를 더 쓰고, ObjectMapper 등의 객체 생성과 스레드 내에서 동작하는 직렬화는 추가됐지만, 실질적인 데이터 전송 시에 발생하는 직렬화에 대한 메모리 및 응답 지연 속도 최적화는 되지 않은 것입니다.
public class NetworkMessageDto {
private Long receiverId;
private String eventId;
private String eventName;
private String message;
@JsonRawValue
private String data;
...
}
Jackson에서 확실하게 Json 직렬화를 하지 않게 하기 위해서 데이터 컬럼에 @JsonRawValue를 붙여서 다시 진행해보겠습니다.
211ms로 latency가 감소했습니다. 코드를 수정하기 전의 값인 236ms보다 25ms 더 빨라진 것을 확인할 수 있습니다. queueing delay가 누적되어서 max가 치솟았던 점도 많이 개선됐습니다.
다만 왜 메모리 사용량이 더 늘었는지 모르겠습니다. GC가 5번에서 6번으로 늘어났는데 제 생각대로였다면 줄었어야 했는데 말이죠. Eden의 총량 자체는 minor gc에 대한 처리를 받는데 minor gc는 그렇게 길게 걸리지 않으므로 괜찮다고 생각은 듭니다만, 기존의 제 생각과 다르다는 점이 의아했습니다. Old가 늘어난 것이 아니므로 일단 접어두겠습니다.
기대했던 CPU 사용량인데요, 튀는 부분없이 평이하게 흘러가고 있습니다. 코드 설계에서 의도한 점과 부합합니다.
결론적으로 CPU 사용량 개선과 응답 지연 시간 개선이 잘 이루어진 것 같습니다. 그래도 아직 해결해야할 부분인 최댓값과 최솟값의 너무 큰 오차가 발생하는 것을 개선해보겠습니다. 아래는 다시 살펴보는 send 로직의 코드입니다.
@Override
public void send(List<NetworkMessageDto> messages) {
for (NetworkMessageDto message : messages) {
Long receiverId = message.getReceiverId();
Map<String, SseEmitter> userEmitters = emitterRepository.findAllStartWithById(receiverId);
userEmitters.forEach((emitterId, emitter) -> {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
});
}
}
List로 들어온 NetworkMessageDto를 Loop를 돌면서 하나하나 send합니다. 여기서 생기는 계산 오버헤드를 위에서 처리했었습니다. 다만 이렇게 send를 하게 된다면 List의 내용을 serial하게 처리하게 됩니다. 이렇게 되면 throughput이 현저하게 떨어지고, 특히 처음 sse 이벤트 메시지를 받은 사용자와 마지막에 받은 사용자의 시간 차이는 List가 길어질수록 심해집니다.
위 시퀸스 다이어그램은 직렬 처리 방식으로 보냈을 경우의 처리 방식입니다. N개를 처리할 때 1부터 순차적으로 로직을 처리하게 된다면 N번째 있는 처리 태스크의 경우는 N-1개의 태스크를 처리하는 시간만큼 반드시 기다려야 합니다. 이렇게 되면 마지막 User 간 응답 지연 시간의 격차가 발생하고 점점 차이가 커짐과 동시에 마지막 인원은 실시간 메시지라는 느낌도 받지 못할 수 있습니다.
Virtual thread
응답 지연 시간 개선이 아니라, 단위 시간당 처리할 수 있는 처리량을 늘려야합니다. 여기서 사용할 수 있는 방법이 자바 21부터 제공하고 있는 가상 스레드 입니다. 저렴한 비용의 경량 스레드를 사용자 수만큼 생성하면서 동시에 요청을 보냅니다. 루프 안에서 기다릴 필요 없이 거의 동시 처리에 가까운 수준으로 병렬 전송을 진행할 수 있습니다.
N개의 Emitter에게 send를 하게되면 N개의 스레드를 생성해야하는데, 가상 스레드의 경우 그 비용이 매우 저렴해서 오버헤드가 매우 작습니다.
가설 (2)
가상 스레드로 Loop를 돌면서 send를 진행하던 것을 가상 스레드를 이용해서 스레드 단위 별로 send를 처리한다면 유저 간 응답 지연 속도 차이를 크게 개선할 수 있을 것입니다.
가설 (2) 개선 시도
가상 스레드 풀을 하나 선언하고 Loop 안에서 sendBySse 메서드를 위임하고 즉시 리턴합니다.
private final ExecutorService sseExecutor = Executors.newVirtualThreadPerTaskExecutor();
@Override
public void send(List<NetworkMessageDto> messages) {
for (NetworkMessageDto message : messages) {
Long receiverId = message.getReceiverId();
Map<String, SseEmitter> userEmitters = emitterRepository.findAllStartWithById(receiverId);
userEmitters.forEach((emitterId, emitter) -> {
sseExecutor.submit(() -> {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
});
});
}
}
이렇게 진행하면 현재의 경우 경량 스레드가 1만 개가 빠르게 생성됩니다. 가상 스레드의 경우 생성 시간이 1μs이기 때문에 추가적으로 생기는 오버헤드는 1μs * 10000 = 10ms입니다.
가설 (2) 개선 결과
평균 응답 지연 시간이 상당히 개선됐습니다. 211ms에서 171ms로 줄었습니다. 최댓값이 1396ms에서 496ms로 줄어서 결과적으로 평균 값이 줄어들은 결과로 보입니다. 응답 지연시간이 전체적으로 의도된 대로 많이 개선된 모습을 확인할 수 있습니다.
CPU 사용량입니다. 0.1이 안되는 정도, 0.05 정도로 일정하게 유지되고 있습니다. 스파이크 없이 잘 돌아가는 모습을 확인할 수 있었습니다. 이것도 이전의 결과에서 특별한 추가 오버헤드 없이 개선된 상태로 잘 유지됨을 확인했습니다.
결과적으로 C1K에서 스레시 홀드 내로 결과가 수용되면서 잘 돌아가는 것을 확인했습니다.
시나리오 2: 10,000명의 유저에게 동시에 SSE 메시지를 전송한다.
이제 C10K 상황으로 들어가보겠습니다. 시나리오를 조금 수정해야할 것 같습니다. 2분 동안 1천명 연결에서 3분 동안 1만명 연결로, 4분 동안 유지로 설정을 바꾸었고 알림 전송 로직 실행도 5초마다 1번씩으로 수정했습니다.
export const options = {
scenarios: {
// (Subscribers)
audience: {
executor: 'ramping-vus',
exec: 'audience',
startVUs: 0,
stages: [
{ duration: '3m', target: 10000 }, // 3분 동안 1만명 연결
{ duration: '4m', target: 10000 }, // 4분 동안 유지
],
gracefulStop: '30s',
},
director: {
executor: 'constant-arrival-rate',
exec: 'director',
rate: 1,
timeUnit: '5s',
duration: '5m',
preAllocatedVUs: 1,
startTime: '2m30s',
},
},
};테스트 결과
일단 1만명 받기에는 성공했습니다. fd가 1만개 만큼 상승했기 때문이죠. 이전에 1만명을 받은 적이 있으니, 이 부분은 크게 걱정되지는 않았습니다.
메모리 사용량을 보면, 여러 번의 minor gc 끝에 Old가 1.33GB만큼 사용량이 올라갔습니다. 기존 사용량이 1MB니까 무시할만 해서 SSE 구독으로 인해 사용된 Old 메모리 양이 1.33GB라고 생각해도 될 것 같습니다. 1330MB/10000 = 0.133MB = 133KB입니다. 1명의 구독을 수용하는데 133KB입니다. 이전에는 169MB/1000 = 169KB입니다. 이는 SSE Emitter 객체 뿐만 아니라, 톰캣이 연결 하나를 유지하기 위한 Buffer 크기가 포함되어 있기 때문에 나오는 수치입니다. 톰캣은 기본 연결 하나당 입/출력 버퍼를 각각 16KB를 할당하고, 서블릿 요청, 응답 객체가 합쳐서 10KB 정도를 차지합니다. 스레드 로컬의 컨텍스트 정보 등까지 스프링과 톰캣 오버헤드를 합쳐서 상정하면 100KB정도는 나오기 때문에 괜찮은 수치로 보입니다. Old 메모리를 최대 6기가를 할당해줬기 때문에 버틸 수 있는 최대 용량은 단순 계산으로 4GB로 생각됩니다. 아마 3GB가 최대가 되지 않을까 생각합니다. 아무튼 메모리 사용량은 준수해보입니다.
망했습니다...
평균 응답 지연 시간이 55444ms, 즉 55초가 나왔습니다. 최소가 29이고 최대가 126초, 약 2분이 걸립니다. 중간에 큰 병목이 생긴 것 같습니다. 다만 이 정도의 큰 병목은 비즈니스 로직에서 발생했다고 보기에는 너무 크다고 생각이 듭니다.
CPU가 받아주지 못하고 있어서라는 생각도 듭니다. 최적화를 한다고 하긴 했으나 JVM에서 0.1 만큼도 쓰지 않았을 때에도 System CPU가 최대 0.5까지는 썼었는데, 지금은 1을 계속해서 찍고있으니 말이죠. 그래도 수치적으로는 간당간당하게 되어야 하지 않나,하고 생각은 듭니다.
WAS에서 sse를 전송을 완료하고 클라이언트에서 받는 데까지 걸린 시간이 WAS에서 비즈니스 로직 처리 시간이 오래 걸린 것인지, 아니면 컴퓨터 내부 부하가 심하게 와서 K6에만 늦게 도착한 것인지 잘 분간이 안돼서 순수하게 스레드가 send를 끝낸 시간만 측정해보려고 합니다.
@Override
public void send(List<NetworkMessageDto> messages) {
long startTime = System.currentTimeMillis();
int totalCount = messages.size();
CountDownLatch latch = new CountDownLatch(totalCount);
for (NetworkMessageDto message : messages) {
Long receiverId = message.getReceiverId();
Map<String, SseEmitter> userEmitters = emitterRepository.findAllStartWithById(receiverId);
userEmitters.forEach((emitterId, emitter) -> {
sseExecutor.submit(() -> {
try {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
} finally {
latch.countDown();
}
});
});
try {
boolean completed = latch.await(10, TimeUnit.SECONDS);
if (!completed) {
log.warn("일부 전송이 시간 내에 완료되지 않았습니다.");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
long endTime = System.currentTimeMillis();
log.info("전송 완료 소요 시간: {}ms", (endTime - startTime));
}
}
이렇게 코드를 구성해서, 실제로 얼마나 걸렸는지 체크해보도록 하겠습니다. 커널의 Send-Q에 이상이 있는지 점검해보기 위해서 다음과 같은 코드를 터미널에 입력해서 모니터링했습니다.
while true; do
clear
date
echo "=== Send-Q(송신 버퍼) 정체 현황 (Top 20) ==="
echo "Proto Recv-Q Send-Q Local Address Foreign Address (state)"
netstat -anp tcp | grep "\.8080 " | awk '$3 > 0' | sort -nr -k 3 | head -n 20
sleep 1
done결과는 아래와 같습니다.
스레드 별로 얼마나 걸리는지도 체크해보겠습니다. 일정 시간이 넘어가면 로그를 남기도록 구성했습니다.
log.info("현재 스레드: {}", Thread.currentThread());
서버 로그
2026-02-13T15:44:38.559+09:00 WARN 96490 --- [ task-1] .e.B.n.s.n.SseNotificationNetworkService : 일부 전송이 시간 내에 완료되지 않았습니다.
2026-02-13T15:44:38.560+09:00 INFO 96490 --- [ task-1] .e.B.n.s.n.SseNotificationNetworkService : 전송 완료 소요 시간: 192332ms
2026-02-13T15:44:38.617+09:00 INFO 96490 --- [ virtual-207] .e.B.n.s.n.SseNotificationNetworkService : 현재 스레드: VirtualThread[#207]/runnable@ForkJoinPool-1-worker-1
가상 스레드의 호출은 잘 되고 있고, 소켓 Send-Q가 계속 0으로 관찰되고 있음을 보아서 커널 소켓 레벨에서는 문제가 없음을 확인했습니다.
혹시 가상 스레드가 pinned 현상이 나타나고 있나 확인해보려고 VM Options를 수정해서 확인해보려고 했으나, 해당 로그를 확인하지는 못 했습니다. 사용자 수에 의해서 터지는 지에 대해서 알고싶어서 3천명일 때의 상황에서도 점검해봤으나, 같은 문제가 야기됨을 확인했습니다.
Send-Q도 잘 비워지고 있었고, DB에 대한 문제는 사전에 해결한 바 있습니다. 가상 스레드로 변경을 해서 throughput에 대해서도 최적화를 했는데 이런 상황이 발생한다면, 전적으로 send 로직의 시간 복잡도의 문제라고 판단했습니다. 아래는 구현한 send 메서드입니다.
@Override
public void send(List<NetworkMessageDto> messages) {
for (NetworkMessageDto message : messages) { // O(n)
Long receiverId = message.getReceiverId();
Map<String, SseEmitter> userEmitters = emitterRepository.findAllStartWithById(receiverId); // O(n)
// O(1)
userEmitters.forEach((emitterId, emitter) -> {
sseExecutor.submit(() -> {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
});
});
}
}여기서 Message 객체를 반복문으로 도는 것은 필수적입니다. 다만 그 내부에서 반복문으로 emitter를 조회하는 로직이 존재합니다. 이 로직까지 포함한다면 O(n^2)으로 제곱에 해당하는 시간복잡도가 형성됩니다. n^2에 대해서는 분명히 해결해야하는 부분이기 때문에 emitterRepository에서 메서드 구현부를 살펴보겠습니다.
@Repository
public class EmitterRepository {
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
public Map<String, SseEmitter> findAllStartWithById(Long memberId) {
Map<String, SseEmitter> result = new ConcurrentHashMap<>();
emitters.forEach((id, emitter) -> {
if(id.startsWith(memberId.toString() + "_")){
if (emitter != null) {
result.put(id, emitter);
}
}
});
return result;
}
}EmitterRepository에서 findAllStartWithById의 구현부입니다. SSE 연결의 특성 상, 하나의 사용자 당 여러 개의 SSE 커넥션이 생길 수 있습니다. 특히 하나의 브라우저 엔진에서 탭을 여러 개 띄우는 행위에서 SSE 연결이 N개가 될 수 있음을 생각하면 Emitter가 사용자 별로 여러 개가 생길 수 있음은 자명합니다. HTTP/1.1에서만 그렇게 되고, HTTP/2는 멀티플렉싱을 지원하니 SSE 연결이 하나인 것이 아니냐라고 생각할 수 있습니다만, 물리적인 TCP 연결을 1개만 생성하고 공유하는 것이지, 각각의 탭마다 Stream 이벤트가 생성되기 때문에 서버 입장에서는 이 스트림 하나하나가 새로운 HTTP 요청으로 인식됩니다. 따라서 탭 개수에 따라서 Emitter가 N개 생성됩니다. 핸드폰과 데스크탑에서 동시에 로그인 할 경우에도 마찬가지로 복수의 SSE Emitter가 사용됩니다.
사용자당 SSE 연결이 복수로 생길 수 있음은 피할 수 없는 것 같습니다. 다만 Send 로직 내에서 사용자별로 생성된 복수의 Emitter를 찾기 위해서 Map을 Full scan해서 찾아오는 방식은 비효율적으로 보입니다. 이를 개선해서 Send 로직을 O(n^2)에서 O(n)으로 개선해보겠습니다.
Nested Map Structure (Bucketing)
지금은 모든 Emitter를 하나의 맵 객체에 전부 넣고 있습니다.
Map<String, SseEmitter> emitters --------------------------------
"user1_time1" -> Emitter A
"user1_time2" -> Emitter B
"user2_time1" -> Emitter C ...
(사용자별 SSE Emitter 전체가 섞여 있음, Flat한 구조)
여기서 "사용자별로 다시 Emitter를 묶어서 Map에 보관"한다면 굳이 반복문으로 사용자의 Id를 이용해서 객체를 찾지 않고 Loop 없이 가져올 수 있습니다.
Map<Long, Map<String, SseEmitter>> emitterStorage ------------------------------------------------- Key(user1) -> { "user1_time1": Emitter A, "user1_time2": Emitter B }
Key(user2) -> { "user2_time1": Emitter C } ...
(유저별로 그룹화됨, Bucketing)
Emitter를 보관하는 자료구조를 Bucketing해서 운용한다면 시간복잡도를 줄일 수 있을 것입니다.
가설 (1)
Emitter를 조회하는 데 불필요한 시간복잡도가 부여돼있습니다. 이를 해결하면 send 로직 자체가 빨라져서 응답 지연시간을 개선할 수 있을 것입니다.
가설 (1) 개선 시도
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
public SseEmitter save(String emitterId, SseEmitter emitter) {
emitters.put(emitterId, emitter);
return emitter;
}기존의 코드입니다. 이를 사용자 단위로 다시 Map 객체로 묶는 자료구조로 개선해보겠습니다.
private final Map<Long, Map<String, SseEmitter>> emitterStorage = new ConcurrentHashMap<>();
public SseEmitter save(Long memberId, String emitterId, SseEmitter emitter) {
emitterStorage.computeIfAbsent(memberId, key -> new ConcurrentHashMap<>())
.put(emitterId, emitter);
return emitter;
}
public Map<String, SseEmitter> findAllStartWithById(Long memberId) {
Map<String, SseEmitter> userEmitters = emitterStorage.get(memberId);
return userEmitters != null ? userEmitters : Collections.emptyMap();
}기존의 자료구조를 개선해서 memberId를 기준으로 bucketing했습니다. 이제 findAllStartWithById()가 O(n)에서 O(1)의 시간복잡도로 개선돼서, 기존의 로직이 O(n)의 시간복잡도로 개선됐습니다. 3천명으로 테스트해보겠습니다.
가설 (1) 개선 결과
max값이 961로 많은 개선을 보였습니다. p(95)도 700ms 정도로 많은 개선을 보였습니다.
다만 메모리 사용량은 더 높아졌습니다. 이유는 당연히도 맵 구조가 중첩되었기 때문입니다. 다만 이는 필요한 오버헤드라고 볼 수 있습니다. 다시 원래 문제로 돌아가서 1만명의 경우를 다시 테스트해보겠습니다!
2중 맵 객체를 사용했는데도 기존과 비교했을 때, 그렇게 큰 메모리 사용량을 사용하지 않는것 같습니다. N명의 사용자의 경우에 N개 만큼의 객체가 더 생성돼서 Old 영역이 튈까봐 걱정했는데, 매우 고무적입니다. Map 객체 자체는 그렇게 큰 메모리를 사용하지는 않는 것 같습니다.
CPU도 시스템 사용량에 비해 매우 안정적입니다. 이전 처럼 튀는 현상도 없고, 최댓값이 0.06인 것을 보니 계산량 자체도 안정적입니다.
1만명의 부하를 걸었을 때, 확실하게 이전보다 훨씬 많은 개선을 보였습니다. 평균이 55초에서 1.4초로 개선됐으니까요. 이상하리만치 벗어나있는 max는 아직 15초입니다. 근데 이전에는 p(95)값도 매우 컸었는데, 지금은 p(95)값은 2665ms로 비교적 안정적입니다. 그런데 max값은 15초로 p(95)에 비해서도 매우 큽니다.
G1의 Eden 영역이 꽉차서 GC가 일어났다고 해도, minor GC이기 때문에 매우 빨라서 Stop-The-World가 최대 33ms 밖에 일어나지 않았습니다. 또한 virtual thread의 pinning 현상도 로그로 관찰할 수 없었습니다. 따라서 send-Q에 대해서 모니터링할 필요를 느꼈습니다. 다시 테스트를 돌려보고, send-Q 버퍼가 네트워크로 전송하는 속도보다 쌓이는 속도가 더 빨라서 쌓이는지 체크해보겠습니다. 위와 같은 명령어로 모니터링해보겠습니다.
커널의 send-Q는 최대 145 정도로 쌓이다가 1초 뒤에 사라지는 모습을 보였습니다. 이로써 send-Q가 쌓이지 않고 잘 전송돼고 있는 것을 확인했습니다. 그렇다면 과하게 p(95)의 범위를 과하게 벗어나는 것은 로직에서 에러가 났는데, 클라이언트가 매우 긴 timeout과 함께 기다린 몇 개의 요청에서만 걸린 시간으로 보여집니다. 본 서비스에서 실시간 sse 메시지는 단순히 알림이 도착했다고만 알려주는 로직이므로 p(95) 환경에서 적절한 값을 관측했으므로 테스트는 성공이라고 판단됩니다. 기존의 1초의 스레시 홀드 기준에서는 실패이지만, 말이죠.
Fail-Fast 전략
2분으로 찍혀서 의미없는 기다림을 길게 가져갈 바에, 차라리 빠르게 Fail시키고 에러를 뱉는것이 시스템에 있어서 더 안정성을 높일 것 같습니다. 따라서 애플리케이션 레벨의 timeout을 선언해서 max값을 최소화하는 전략을 취해보겠습니다.
userEmitters.forEach((emitterId, emitter) -> {
CompletableFuture.runAsync(() -> {
sendBySse(
emitter,
emitterId,
message.getEventId(),
message.getEventName(),
message.getData()
);
}, sseExecutor)
.orTimeout(3, TimeUnit.SECONDS)
.exceptionally(ex -> {
emitter.completeWithError(ex);
emitterRepository.deleteById(emitterId);
return null;
});
});이렇게 하면 해당 가상 스레드에서 3초 동안 해당 작업을 수행하지 못하면 timeout이 터지면서 해당 작업은 실패한 것으로 간주합니다. 해당 sse emitter를 메모리에서 정리합니다. 스레드 별로 timeout이 생기므로 해당 에미터에 대해서만 timeout이 발생해서 나머지는 건강하게 돌아갑니다. 이때 연결이 끊기면 클라이언트에서는 SSE 특성 상 자동으로 재연결을 시도하므로 다시 연결될 것입니다. 다시 연결이 돼면 그 알림 자체에 대해서는 유실되지만, Last-Event-Id를 기반으로 한 재접속 시 정합성을 위한 정책도 마련이 돼있기 때문에 이 정도 처리로 충분할 것입니다.
모든 태스크를 완벽하게 처리했으므로 예외처리는 발생하지 않았고 JVM에서 태스크를 포기하는 타임아웃이 3초이지, 클라이언트 측에서 타임아웃을 설정한 것이 아니기 때문에 max는 36초입니다. 서버의 문제를 떠난 것 같습니다.
C10K 이상의 부하
기존에 3만명에 대해서 SSE Emitter 객체의 연결에 대해서 부하가 심해서 메모리도 CPU도 간당간당하게 돌아갔었습니다. 메모리 관점에서도 그렇지만 분명 속도 관점에서도 대상이 늘면 늘수록 한계치에 다다를 것입니다. 따라서 메모리와 응답 지연 속도의 관점에서 C30K에서 WebFlux로 로직을 전환해서 모놀리스 서버에서 최대한 개선을 해보도록 하겠습니다. 해당 포스트는 이어서 다음 포스트에서 작성해보도록 하겠습니다!
'Test' 카테고리의 다른 글
| [Test] 점진적 부하테스트: 50,000명 유저의 SSE 구독 유지 (1) | 2026.03.03 |
|---|---|
| [Test] 점진적 부하테스트: 20,000명 유저의 SSE 메시지 전송 - WebFlux (2) | 2026.02.20 |
| [Test] 점진적 부하테스트: 1,000,000명 유저의 알림 전송 요청 - 스레드 (1) | 2026.02.10 |
| [Test] 점진적 부하테스트: 10,000명 유저의 알림 전송 요청 - JPA (0) | 2026.02.06 |
| [Test] 점진적 부하테스트: 30,000명 유저의 SSE 구독 요청 - TCP (0) | 2026.02.04 |