diff --git a/result-bulkhead-before.json b/result-bulkhead-before.json new file mode 100644 index 00000000..7aeddc6b --- /dev/null +++ b/result-bulkhead-before.json @@ -0,0 +1,197 @@ +{ + "root_group": { + "name": "", + "path": "", + "id": "d41d8cd98f00b204e9800998ecf8427e", + "groups": {}, + "checks": { + "[notif] status 200": { + "passes": 1567, + "fails": 0, + "name": "[notif] status 200", + "path": "::[notif] status 200", + "id": "9fc740afe9cf21a0327678441a348978" + }, + "[ws] status 101": { + "name": "[ws] status 101", + "path": "::[ws] status 101", + "id": "3cc6802624f263d3928ed6de98f3c352", + "passes": 2, + "fails": 0 + } + } + }, + "metrics": { + "ws_connecting": { + "max": 1155.4575, + "p(90)": 1155.34769, + "p(95)": 1155.402595, + "avg": 1154.90845, + "min": 1154.3594, + "med": 1154.90845 + }, + "ws_msgs_sent": { + "count": 1651, + "rate": 4.985553421339417 + }, + "ws_msgs_received": { + "count": 3304, + "rate": 9.977146277471492 + }, + "http_reqs": { + "count": 1567, + "rate": 4.731897160047769 + }, + "http_req_receiving": { + "p(90)": 0.7683200000000002, + "p(95)": 0.9485099999999989, + "avg": 0.21188544990427544, + "min": 0, + "med": 0, + "max": 6.9107 + }, + "checks": { + "passes": 1569, + "fails": 0, + "value": 1 + }, + "vus_max": { + "value": 12, + "min": 12, + "max": 12 + }, + "chat_messages_received": { + "count": 1648, + "rate": 4.97649426915043 + }, + "iterations": { + "count": 1569, + "rate": 4.737936594840427 + }, + "ws_sessions": { + "count": 2, + "rate": 0.006039434792658289 + }, + "http_req_duration{expected_response:true}": { + "p(90)": 2076.8168400000004, + "p(95)": 2721.69289, + "avg": 1151.817254243779, + "min": 431.0899, + "med": 977.7864, + "max": 4974.7035 + }, + "http_req_blocked": { + "p(90)": 0, + "p(95)": 0, + "avg": 1.902208615188258, + "min": 0, + "med": 0, + "max": 315.6305 + }, + "http_req_connecting": { + "max": 148.6131, + "p(90)": 0, + "p(95)": 0, + "avg": 0.895541671984684, + "min": 0, + "med": 0 + }, + "chat_messages_sent": { + "count": 1649, + "rate": 4.979513986546759 + }, + "http_req_sending": { + "med": 0, + "max": 3.377, + "p(90)": 0.6262200000000001, + "p(95)": 0.7184799999999996, + "avg": 0.17865022335673258, + "min": 0 + }, + "vus": { + "value": 2, + "min": 2, + "max": 12 + }, + "data_sent": { + "count": 251316, + "rate": 758.9032971758552 + }, + "http_req_waiting": { + "p(95)": 2721.5110299999997, + "avg": 1151.4267185705173, + "min": 430.2286, + "med": 977.2512, + "max": 4974.1908, + "p(90)": 2076.08822 + }, + "chat_e2e_latency": { + "min": 350, + "med": 404, + "max": 5422, + "p(90)": 3371.499999999998, + "p(95)": 4390.65, + "avg": 937.6231796116505, + "thresholds": { + "p(95)<3000": true + } + }, + "notif_req_duration": { + "med": 977.7864, + "max": 4974.7035, + "p(90)": 2076.8168400000004, + "p(95)": 2721.69289, + "avg": 1151.817254243779, + "min": 431.0899 + }, + "ws_session_duration": { + "min": 331155.5802, + "med": 331155.89955, + "max": 331156.2189, + "p(90)": 331156.15502999997, + "p(95)": 331156.186965, + "avg": 331155.89955 + }, + "notif_errors": { + "passes": 0, + "fails": 1567, + "thresholds": { + "rate<0.10": false + }, + "value": 0 + }, + "http_req_failed": { + "fails": 1567, + "passes": 0, + "value": 0 + }, + "iteration_duration": { + "avg": 1574.7665509878918, + "min": 431.6325, + "med": 982.5982, + "max": 331156.2189, + "p(90)": 2088.631680000001, + "p(95)": 2724.6312 + }, + "http_req_duration": { + "p(90)": 2076.8168400000004, + "p(95)": 2721.69289, + "avg": 1151.817254243779, + "min": 431.0899, + "med": 977.7864, + "max": 4974.7035 + }, + "data_received": { + "count": 1517803, + "rate": 4583.336123300564 + }, + "http_req_tls_handshaking": { + "max": 181.0315, + "p(90)": 0, + "p(95)": 0, + "avg": 1.003557562220804, + "min": 0, + "med": 0 + } + } +} \ No newline at end of file diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatMessageSendEventListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatMessageSendEventListener.java index b468ce10..07619792 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatMessageSendEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatMessageSendEventListener.java @@ -16,7 +16,7 @@ public class ChatMessageSendEventListener { private final ChatSendService chatSendService; @EventListener - @Async + @Async("chatExecutor") public void handleChatMessageSend(ChatMessageSendEvent event) { log.info("메시지 전송 이벤트 처리 - 채팅방: {}, 발신자: {}", event.chatRoomId(), event.senderId()); diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatNotificationEventListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatNotificationEventListener.java index 1051d1a7..83b69519 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatNotificationEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatNotificationEventListener.java @@ -17,7 +17,7 @@ public class ChatNotificationEventListener { private final ChatPushNotificationService chatPushNotificationService; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("notificationExecutor") public void handleChatNotification(ChatNotificationEvent event) { log.info("채팅 알림 이벤트 처리 - 채팅방: {}, 발신자: {}", event.chatRoomId(), event.senderId()); try { diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomListUpdateEventListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomListUpdateEventListener.java index 9f324c0a..921ed92a 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomListUpdateEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomListUpdateEventListener.java @@ -24,7 +24,7 @@ public class ChatRoomListUpdateEventListener { private final ChatRoomListUpdateBroadcaster chatRoomListUpdateBroadcaster; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("chatExecutor") public void handleChatRoomListUpdate(ChatRoomListUpdateEvent event) { log.info("채팅방 목록 업데이트 이벤트 처리 시작 - 채팅방: {}", event.chatRoomId()); diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomRedisCleanupListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomRedisCleanupListener.java index 454cec91..26451647 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomRedisCleanupListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatRoomRedisCleanupListener.java @@ -21,7 +21,7 @@ public class ChatRoomRedisCleanupListener { private final ChatListSubscriptionStore chatListSubscriptionStore; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("chatExecutor") public void handleChatRoomRedisCleanup(ChatRoomRedisCleanupEvent event) { Long chatRoomId = event.chatRoomId(); log.info("[채팅방 Redis 정리 시작] - chatRoomId: {}", chatRoomId); diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatSubscriptionEventListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatSubscriptionEventListener.java index 61bb29bb..8bb50713 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatSubscriptionEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatSubscriptionEventListener.java @@ -49,7 +49,7 @@ public void handleChatRoomSubscription(ChatRoomSubscriptionEvent event) { } @EventListener - @Async + @Async("chatExecutor") public void handleChatListSubscription(ChatListSubscriptionEvent event) { log.info("채팅방 목록 구독 이벤트 처리 시작 - 멤버: {}, 액션: {}, 채팅방 수: {}", event.memberId(), event.action(), event.chatRoomIds().size()); diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventListener.java index 4aba4aa3..414704dc 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventListener.java @@ -30,7 +30,7 @@ public class ChatUnreadStatusUpdateEventListener { private final ChatSessionRegistry chatSessionRegistry; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("chatExecutor") public void handleChatUnreadStatusUpdate(ChatUnreadStatusUpdateEvent event) { log.info("채팅 안읽음 상태 업데이트 이벤트 처리 시작 - 대상자: {}명", event.targetMemberIds().size()); diff --git a/src/main/java/umc/cockple/demo/domain/chat/service/listener/PartyChatSystemMessageListener.java b/src/main/java/umc/cockple/demo/domain/chat/service/listener/PartyChatSystemMessageListener.java index ad7a10ea..c88ff845 100644 --- a/src/main/java/umc/cockple/demo/domain/chat/service/listener/PartyChatSystemMessageListener.java +++ b/src/main/java/umc/cockple/demo/domain/chat/service/listener/PartyChatSystemMessageListener.java @@ -15,7 +15,7 @@ public class PartyChatSystemMessageListener { private final ChatSendService chatSendService; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("chatExecutor") public void handlePartyMemberChanged(PartyMemberJoinedEvent event) { switch (event.action()) { case JOINED -> chatSendService.sendSystemMessage(event.partyId(), diff --git a/src/main/java/umc/cockple/demo/domain/notification/events/NotificationEventListener.java b/src/main/java/umc/cockple/demo/domain/notification/events/NotificationEventListener.java index 8af12a45..9b0a772f 100644 --- a/src/main/java/umc/cockple/demo/domain/notification/events/NotificationEventListener.java +++ b/src/main/java/umc/cockple/demo/domain/notification/events/NotificationEventListener.java @@ -21,7 +21,7 @@ public class NotificationEventListener { private final MemberRepository memberRepository; @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) - @Async + @Async("notificationExecutor") @Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW) public void handleNotification(NotificationEvent event) { log.info("[NOTIFICATION] FCM 전송 이벤트 처리 - memberId: {}", event.memberId()); diff --git a/src/main/java/umc/cockple/demo/global/config/AsyncConfig.java b/src/main/java/umc/cockple/demo/global/config/AsyncConfig.java index e9e74d1a..9e77abc8 100644 --- a/src/main/java/umc/cockple/demo/global/config/AsyncConfig.java +++ b/src/main/java/umc/cockple/demo/global/config/AsyncConfig.java @@ -1,5 +1,7 @@ package umc.cockple.demo.global.config; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration; import org.springframework.boot.task.ThreadPoolTaskExecutorBuilder; import org.springframework.context.annotation.Bean; @@ -9,16 +11,32 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import umc.cockple.demo.global.logging.MdcTaskDecorator; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadPoolExecutor; + +/** + * 지연 특성별로 벌크헤드(bulkhead) 격리 + * {@code chatExecutor} — 실시간/저지연 + * {@code notificationExecutor} — 외부 FCM HTTP 호출 + * {@code applicationTaskExecutor} — qualifier 없는 @Async의 안전한 기본 착지점 + */ @Configuration @EnableAsync +@Slf4j public class AsyncConfig { + /** Spring이 qualifier 없는 @Async의 기본 executor를 찾는 이름. */ + private static final String DEFAULT_TASK_EXECUTOR_BEAN_NAME = "taskExecutor"; + @Bean public TaskDecorator mdcTaskDecorator() { return new MdcTaskDecorator(); } - @Bean(name = TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME) + @Bean(name = { + TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME, + DEFAULT_TASK_EXECUTOR_BEAN_NAME + }) public ThreadPoolTaskExecutor applicationTaskExecutor( ThreadPoolTaskExecutorBuilder builder, TaskDecorator mdcTaskDecorator @@ -28,4 +46,65 @@ public ThreadPoolTaskExecutor applicationTaskExecutor( .taskDecorator(mdcTaskDecorator) .build(); } + + /** + * 실시간 채팅용 풀-> 짧은 내부 I/O 작업을 저지연으로 처리 + */ + @Bean("chatExecutor") + public ThreadPoolTaskExecutor chatExecutor( + @Value("${async.chat.core-size:8}") int coreSize, + @Value("${async.chat.max-size:16}") int maxSize, + @Value("${async.chat.queue-capacity:500}") int queueCapacity, + @Value("${async.chat.await-termination-seconds:30}") int awaitTerminationSeconds, + TaskDecorator mdcTaskDecorator + ) { + // 채팅은 유실 방지가 우선 -> 포화 시 호출 스레드에서 실행해 백프레셔를 건다. + return buildExecutor("cockple-chat-", coreSize, maxSize, queueCapacity, + awaitTerminationSeconds, mdcTaskDecorator, new ThreadPoolExecutor.CallerRunsPolicy()); + } + + /** + * FCM 알림용 풀. 외부 Firebase HTTP 호출을 채팅과 격리 + * 큐를 크게 잡되 유한하게 두어 Firebase 장애 시 무한 백로그(OOM)를 방지 + */ + @Bean("notificationExecutor") + public ThreadPoolTaskExecutor notificationExecutor( + @Value("${async.notification.core-size:4}") int coreSize, + @Value("${async.notification.max-size:12}") int maxSize, + @Value("${async.notification.queue-capacity:1000}") int queueCapacity, + @Value("${async.notification.await-termination-seconds:30}") int awaitTerminationSeconds, + TaskDecorator mdcTaskDecorator + ) { + + return buildExecutor("cockple-noti-", coreSize, maxSize, queueCapacity, + awaitTerminationSeconds, mdcTaskDecorator, logAndDiscardPolicy()); + } + + /** + * 큐/스레드 포화로 거부된 작업을 예외 전파 없이 버림 + */ + private RejectedExecutionHandler logAndDiscardPolicy() { + // TODO: 알림 재전송 도입 시, 여기서 drop 대신 outbox/DLQ에 적재해 재시도 훅으로 사용 + return (runnable, executor) -> + log.warn("[NOTIFICATION] 알림 풀 포화로 작업 drop - poolSize={}, activeCount={}, queueSize={}", + executor.getPoolSize(), executor.getActiveCount(), executor.getQueue().size()); + } + + private ThreadPoolTaskExecutor buildExecutor(String threadNamePrefix, int coreSize, int maxSize, + int queueCapacity, int awaitTerminationSeconds, + TaskDecorator mdcTaskDecorator, + RejectedExecutionHandler rejectedExecutionHandler) { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(coreSize); + executor.setMaxPoolSize(maxSize); + executor.setQueueCapacity(queueCapacity); + executor.setThreadNamePrefix(threadNamePrefix); + executor.setTaskDecorator(mdcTaskDecorator); + executor.setRejectedExecutionHandler(rejectedExecutionHandler); + // graceful shutdown: 진행 중인 작업이 끝날 때까지 대기 + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(awaitTerminationSeconds); + // 스프링이 빈 초기화 시 afterPropertiesSet()으로 initialize()를 호출한다. + return executor; + } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 903b6e69..566ea401 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -85,6 +85,20 @@ jwt: access-token-validity: 900000 refresh-token-validity: 1209600000 +async: + # 실시간 채팅 풀 (저지연 내부 I/O) + chat: + core-size: 8 + max-size: 16 + queue-capacity: 500 + await-termination-seconds: 30 + # FCM 알림 풀 + notification: + core-size: 4 + max-size: 12 + queue-capacity: 1000 + await-termination-seconds: 30 + fcm: fake-latency-ms: 0 diff --git a/src/test/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventFlowTest.java b/src/test/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventFlowTest.java index f82b0821..721d2136 100644 --- a/src/test/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventFlowTest.java +++ b/src/test/java/umc/cockple/demo/domain/chat/service/listener/ChatUnreadStatusUpdateEventFlowTest.java @@ -7,9 +7,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Primary; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; +import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.bean.override.mockito.MockitoBean; import org.springframework.web.socket.WebSocketSession; import umc.cockple.demo.domain.chat.domain.ChatMessage; @@ -50,14 +50,16 @@ import static org.mockito.Mockito.times; @DisplayName("ChatUnreadStatusUpdateEvent after-commit flow") +// 채팅 @Async 풀(chatExecutor)을 SyncTaskExecutor로 덮어써 동기 실행하려면 빈 오버라이드 허용 필요 +@TestPropertySource(properties = "spring.main.allow-bean-definition-overriding=true") class ChatUnreadStatusUpdateEventFlowTest extends IntegrationTestBase { + // handleChatUnreadStatusUpdate의 @Async("chatExecutor")를 동기로 실행해 테스트 타이밍 문제를 제거 @TestConfiguration static class SyncAsyncConfig { @Bean - @Primary - public TaskExecutor taskExecutor() { + public TaskExecutor chatExecutor() { return new SyncTaskExecutor(); } } diff --git a/src/test/java/umc/cockple/demo/domain/notification/events/NotificationEventListenerTest.java b/src/test/java/umc/cockple/demo/domain/notification/events/NotificationEventListenerTest.java index acfff275..9d8edc11 100644 --- a/src/test/java/umc/cockple/demo/domain/notification/events/NotificationEventListenerTest.java +++ b/src/test/java/umc/cockple/demo/domain/notification/events/NotificationEventListenerTest.java @@ -8,9 +8,9 @@ import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Primary; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; +import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.bean.override.mockito.MockitoBean; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.support.TransactionTemplate; @@ -27,14 +27,15 @@ import static org.mockito.Mockito.never; @DisplayName("NotificationEventListener") +// 알림 @Async 풀(notificationExecutor)을 SyncTaskExecutor로 덮어써 동기 실행하려면 빈 오버라이드 허용 필요 +@TestPropertySource(properties = "spring.main.allow-bean-definition-overriding=true") class NotificationEventListenerTest extends IntegrationTestBase { - // @Async를 동기로 실행해 테스트 스레드와 타이밍 문제를 제거 + // handleNotification의 @Async("notificationExecutor")를 동기로 실행해 테스트 타이밍 문제를 제거 @TestConfiguration static class SyncAsyncConfig { @Bean - @Primary - public TaskExecutor taskExecutor() { + public TaskExecutor notificationExecutor() { return new SyncTaskExecutor(); } } diff --git a/src/test/java/umc/cockple/demo/global/config/AsyncConfigTest.java b/src/test/java/umc/cockple/demo/global/config/AsyncConfigTest.java index 0a248dae..a725ba7a 100644 --- a/src/test/java/umc/cockple/demo/global/config/AsyncConfigTest.java +++ b/src/test/java/umc/cockple/demo/global/config/AsyncConfigTest.java @@ -16,6 +16,8 @@ class AsyncConfigTest { + private final AsyncConfig config = new AsyncConfig(); + @AfterEach void tearDown() { MDC.clear(); @@ -24,17 +26,38 @@ void tearDown() { @Test @DisplayName("applicationTaskExecutor는 MDC를 비동기 작업에 전파하고 다음 작업으로 누수하지 않는다") void applicationTaskExecutorPropagatesMdcAndPreventsLeak() throws Exception { - AsyncConfig config = new AsyncConfig(); - TaskDecorator taskDecorator = config.mdcTaskDecorator(); ThreadPoolTaskExecutor executor = config.applicationTaskExecutor( new ThreadPoolTaskExecutorBuilder() .corePoolSize(1) .maxPoolSize(1) .queueCapacity(1), - taskDecorator + config.mdcTaskDecorator() ); - executor.initialize(); + assertPropagatesMdcAndPreventsLeak(executor); + } + + @Test + @DisplayName("chatExecutor는 MDC를 비동기 작업에 전파하고 다음 작업으로 누수하지 않는다") + void chatExecutorPropagatesMdcAndPreventsLeak() throws Exception { + ThreadPoolTaskExecutor executor = config.chatExecutor(1, 1, 1, 1, config.mdcTaskDecorator()); + + assertPropagatesMdcAndPreventsLeak(executor); + } + + @Test + @DisplayName("notificationExecutor는 MDC를 비동기 작업에 전파하고 다음 작업으로 누수하지 않는다") + void notificationExecutorPropagatesMdcAndPreventsLeak() throws Exception { + ThreadPoolTaskExecutor executor = config.notificationExecutor(1, 1, 1, 1, config.mdcTaskDecorator()); + + assertPropagatesMdcAndPreventsLeak(executor); + } + + /** + * 풀에서 도는 작업에 requestId(MDC)가 전파되고, 스레드 재사용 시 이전 컨텍스트가 누수되지 않는지 검증한다. + */ + private void assertPropagatesMdcAndPreventsLeak(ThreadPoolTaskExecutor executor) throws Exception { + executor.initialize(); try { MDC.put(MdcLoggingFilter.REQUEST_ID, "async-request"); CompletableFuture firstTask = new CompletableFuture<>();