diff --git a/src/main/java/com/matzip/admin/event/RequestReviewEventListener.java b/src/main/java/com/matzip/admin/event/RequestReviewEventListener.java index e93d9fe..7f2af74 100644 --- a/src/main/java/com/matzip/admin/event/RequestReviewEventListener.java +++ b/src/main/java/com/matzip/admin/event/RequestReviewEventListener.java @@ -3,6 +3,7 @@ import com.matzip.common.analytics.AnalyticsRecorder; import com.matzip.common.analytics.domain.EventType; import com.matzip.common.analytics.domain.TargetType; +import com.matzip.common.config.AsyncConfiguration; import com.matzip.common.exception.BusinessException; import com.matzip.common.exception.code.ErrorCode; import com.matzip.lottery.service.LotteryEventService; @@ -28,7 +29,7 @@ public RequestReviewEventListener(LotteryEventService lotteryEventService, Place this.analyticsRecorder = analyticsRecorder; } - @Async + @Async(AsyncConfiguration.GENERAL_EXECUTOR) @TransactionalEventListener(condition = "#event.reviewStatus().name() == 'APPROVED'") public void enterCurrentEventOnApproval(RequestReviewEvent event) { Place place = placeRepository.findById(event.placeId()) diff --git a/src/main/java/com/matzip/common/analytics/AnalyticsEventListener.java b/src/main/java/com/matzip/common/analytics/AnalyticsEventListener.java index c5c3664..e65dc67 100644 --- a/src/main/java/com/matzip/common/analytics/AnalyticsEventListener.java +++ b/src/main/java/com/matzip/common/analytics/AnalyticsEventListener.java @@ -2,6 +2,7 @@ import com.matzip.common.analytics.domain.entity.EventLog; import com.matzip.common.analytics.repository.EventLogRepository; +import com.matzip.common.config.AsyncConfiguration; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; @@ -24,7 +25,7 @@ public AnalyticsEventListener(EventLogRepository eventLogRepository) { this.eventLogRepository = eventLogRepository; } - @Async + @Async(AsyncConfiguration.GENERAL_EXECUTOR) @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) public void on(AnalyticsEvent event) { try { diff --git a/src/main/java/com/matzip/common/config/AsyncConfiguration.java b/src/main/java/com/matzip/common/config/AsyncConfiguration.java index 9a78aee..58b920e 100644 --- a/src/main/java/com/matzip/common/config/AsyncConfiguration.java +++ b/src/main/java/com/matzip/common/config/AsyncConfiguration.java @@ -1,30 +1,64 @@ package com.matzip.common.config; -import org.springframework.boot.task.ThreadPoolTaskExecutorBuilder; +import com.matzip.common.logging.MdcTaskDecorator; +import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.TaskDecorator; import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; - +import java.util.concurrent.RejectedExecutionHandler;@Slf4j @EnableAsync @Configuration public class AsyncConfiguration implements AsyncConfigurer { + public static final String GENERAL_EXECUTOR = "generalExecutor"; + public static final String EXTERNAL_EXECUTOR = "externalExecutor"; + + private static final int SHUTDOWN_AWAIT_SECONDS = 15; + @Override public Executor getAsyncExecutor() { - return asyncTaskExecutor(); + return generalExecutor(); + } + + @Bean(GENERAL_EXECUTOR) + public ThreadPoolTaskExecutor generalExecutor() { + // Hikari 기본 커넥션 풀(10)을 웹 요청과 나눠 쓰도록 max 4로 제한 + return buildExecutor("async-", 3, 4, 200, loggingDiscardPolicy(GENERAL_EXECUTOR)); + } + + @Bean(EXTERNAL_EXECUTOR) + public ThreadPoolTaskExecutor externalExecutor() { + return buildExecutor("ext-", 1, 2, 30, loggingDiscardPolicy(EXTERNAL_EXECUTOR)); + } + + private ThreadPoolTaskExecutor buildExecutor(String threadNamePrefix, int corePoolSize, int maxPoolSize, + int queueCapacity, RejectedExecutionHandler rejectedHandler) { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(corePoolSize); + executor.setMaxPoolSize(maxPoolSize); + executor.setQueueCapacity(queueCapacity); + executor.setThreadNamePrefix(threadNamePrefix); + executor.setRejectedExecutionHandler(rejectedHandler); + executor.setTaskDecorator(mdcTaskDecorator()); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(SHUTDOWN_AWAIT_SECONDS); + executor.initialize(); + return executor; + } + + private TaskDecorator mdcTaskDecorator() { + return new MdcTaskDecorator(); } - @Bean - public TaskExecutor asyncTaskExecutor() { - return new ThreadPoolTaskExecutorBuilder() - .corePoolSize(10) - .maxPoolSize(50) - .queueCapacity(100) - .threadNamePrefix("Async-") - .build(); + // ViewCountService가 REQUIRES_NEW라 CallerRunsPolicy는 커넥션 데드락 위험이 있어 log & discard 사용 + private RejectedExecutionHandler loggingDiscardPolicy(String poolName) { + return (rejected, executor) -> + log.warn("[비동기 작업 유실] {}} 풀 포화로 작업을 버립니다. activeThreads={}, queueSize={}", + poolName, executor.getActiveCount(), executor.getQueue().size()); } } diff --git a/src/main/java/com/matzip/common/infra/discord/DiscordWebhookSender.java b/src/main/java/com/matzip/common/infra/discord/DiscordWebhookSender.java index 0fe8536..9c50fa4 100644 --- a/src/main/java/com/matzip/common/infra/discord/DiscordWebhookSender.java +++ b/src/main/java/com/matzip/common/infra/discord/DiscordWebhookSender.java @@ -1,5 +1,6 @@ package com.matzip.common.infra.discord; +import com.matzip.common.config.AsyncConfiguration; import com.matzip.common.config.DiscordWebhookProperties; import java.util.Map; import lombok.extern.slf4j.Slf4j; @@ -21,7 +22,7 @@ public DiscordWebhookSender(WebClient.Builder webClientBuilder, this.discordWebhookProperties = discordWebhookProperties; } - @Async + @Async(AsyncConfiguration.EXTERNAL_EXECUTOR) public void sendAsync(String content) { String webhookUrl = discordWebhookProperties.getWebhookUrl(); if (!StringUtils.hasText(webhookUrl)) { diff --git a/src/main/java/com/matzip/common/logging/MdcTaskDecorator.java b/src/main/java/com/matzip/common/logging/MdcTaskDecorator.java new file mode 100644 index 0000000..ba84850 --- /dev/null +++ b/src/main/java/com/matzip/common/logging/MdcTaskDecorator.java @@ -0,0 +1,26 @@ +package com.matzip.common.logging; + +import org.slf4j.MDC; +import org.springframework.core.task.TaskDecorator; + +import java.util.Map; + +public class MdcTaskDecorator implements TaskDecorator { + + @Override + public Runnable decorate(Runnable runnable) { + Map parentContext = MDC.getCopyOfContextMap(); + return () -> { + if (parentContext != null) { + MDC.setContextMap(parentContext); + } else { + MDC.clear(); + } + try { + runnable.run(); + } finally { + MDC.clear(); // 풀 스레드 재사용 시 이전 요청 컨텍스트가 남지 않도록 정리 + } + }; + } +} diff --git a/src/main/java/com/matzip/place/service/ViewCountService.java b/src/main/java/com/matzip/place/service/ViewCountService.java index 295d963..a3b1d78 100644 --- a/src/main/java/com/matzip/place/service/ViewCountService.java +++ b/src/main/java/com/matzip/place/service/ViewCountService.java @@ -1,5 +1,6 @@ package com.matzip.place.service; +import com.matzip.common.config.AsyncConfiguration; import com.matzip.place.domain.DailyViewCount; import com.matzip.place.repository.DailyViewCountRepository; import com.matzip.place.repository.PlaceRepository; @@ -18,7 +19,7 @@ public class ViewCountService { private final PlaceRepository placeRepository; private final DailyViewCountRepository dailyViewCountRepository; - @Async + @Async(AsyncConfiguration.GENERAL_EXECUTOR) @Transactional(propagation = Propagation.REQUIRES_NEW) public void incrementAllCounts(Long placeId) { placeRepository.incrementViewCount(placeId); diff --git a/src/test/java/com/matzip/common/config/AsyncConfigurationTest.java b/src/test/java/com/matzip/common/config/AsyncConfigurationTest.java new file mode 100644 index 0000000..fd9890b --- /dev/null +++ b/src/test/java/com/matzip/common/config/AsyncConfigurationTest.java @@ -0,0 +1,37 @@ +package com.matzip.common.config; + +import com.matzip.AbstractMatzipApplicationTest; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +import static org.assertj.core.api.Assertions.assertThat; + +class AsyncConfigurationTest extends AbstractMatzipApplicationTest { + + @Autowired + @Qualifier("generalExecutor") + private ThreadPoolTaskExecutor generalExecutor; + + @Autowired + @Qualifier("externalExecutor") + private ThreadPoolTaskExecutor externalExecutor; + + @Test + void 내부작업_풀과_외부IO_풀이_분리되어_등록된다() { + assertThat(generalExecutor).isNotSameAs(externalExecutor); + } + + @Test + void generalExecutor는_Hikari_커넥션을_고려한_크기로_구성된다() { + assertThat(generalExecutor.getCorePoolSize()).isEqualTo(3); + assertThat(generalExecutor.getMaxPoolSize()).isEqualTo(4); + } + + @Test + void externalExecutor는_저빈도_외부IO용_작은_크기로_구성된다() { + assertThat(externalExecutor.getCorePoolSize()).isEqualTo(1); + assertThat(externalExecutor.getMaxPoolSize()).isEqualTo(2); + } +} diff --git a/src/test/java/com/matzip/common/logging/MdcTaskDecoratorTest.java b/src/test/java/com/matzip/common/logging/MdcTaskDecoratorTest.java new file mode 100644 index 0000000..e6537d4 --- /dev/null +++ b/src/test/java/com/matzip/common/logging/MdcTaskDecoratorTest.java @@ -0,0 +1,68 @@ +package com.matzip.common.logging; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.slf4j.MDC; + +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; + +class MdcTaskDecoratorTest { + + private static final String TRACE_ID = "traceId"; + + private final MdcTaskDecorator decorator = new MdcTaskDecorator(); + + @AfterEach + void clearMdc() { + MDC.clear(); + } + + @Test + void 부모_스레드의_MDC를_실행_스레드로_복사한다() throws Exception { + MDC.put(TRACE_ID, "abc123"); // 제출(부모) 스레드에서 데코레이트 + AtomicReference seenDuringRun = new AtomicReference<>(); + Runnable decorated = decorator.decorate(() -> seenDuringRun.set(MDC.get(TRACE_ID))); + + // 다른 스레드에서 실행 → ThreadLocal이라 데코레이터 없이는 전파되지 않음 + runAndJoin(new Thread(decorated)); + + assertThat(seenDuringRun.get()).isEqualTo("abc123"); + } + + @Test + void 실행_후_스레드의_MDC를_정리한다() throws Exception { + MDC.put(TRACE_ID, "abc123"); + Runnable decorated = decorator.decorate(() -> {}); + + // 실행이 끝난 뒤 워커 스레드에 값이 남으면 재사용 시 다음 작업을 오염시킨다 + AtomicReference afterRun = new AtomicReference<>("SENTINEL"); + runAndJoin(new Thread(() -> { + decorated.run(); + afterRun.set(MDC.get(TRACE_ID)); + })); + + assertThat(afterRun.get()).isNull(); + } + + @Test + void 부모_MDC가_없으면_워커에_남은_잔여값을_지운다() throws Exception { + MDC.clear(); // 부모(제출) 스레드에 컨텍스트 없음 + AtomicReference seenDuringRun = new AtomicReference<>("SENTINEL"); + Runnable decorated = decorator.decorate(() -> seenDuringRun.set(MDC.get(TRACE_ID))); + + // 재사용된 워커 스레드에 이전 작업의 값이 남아있는 상황을 모사 + runAndJoin(new Thread(() -> { + MDC.put(TRACE_ID, "STALE"); + decorated.run(); + })); + + assertThat(seenDuringRun.get()).isNull(); + } + + private void runAndJoin(Thread worker) throws InterruptedException { + worker.start(); + worker.join(); + } +}