Spring Boot Thread & ThreadPool (Feat. @Async, MDC)

안녕하세요 오늘은 Spring Boot의 비동기 처리 방식에 대해서 알아보도록 하겠습니다.

https://velog.io/@gale4739/Spring-Boot-Slack-%EC%A0%81%EC%9A%A9with-ThreadPoolTaskExecutor
저번 게시글에서 ThreadPoolTaskExecutor를 이용하여 에러 로그를 Slack으로 받아보는 작업을 진행했습니다. 이번 시간에는 이 부분에 대해서 좀 더 깊게 알아보는 시간을 갖도록 하겠습니다.

우선 지난 포스팅에서처럼 Slack 메시지 세팅은 기존 스레드에서 진행하고 메시지 발송을 새로운 스레드에서 진행할 경우 스레드들이 서로 컨택스트를 바꿔가며 실행이 되기 때문에 로그가 섞여 로그를 체크하는데 어려움이 있습니다.

이 문제를 Correlation ID를 이용하여 해결할 수 있습니다. 요청마다 고유의 ID(Correlation ID)를 부여하게 되면 아이디를 기준으로 로그를 확인할 수 있습니다.

https://velog.io/@gale4739/Spring-Boot-%EC%9D%B8%EC%A6%9D%EC%9D%B8%EA%B0%80-%ED%94%84%EB%A1%9C%EC%84%B8%EC%8A%A4-%EA%B3%A0%EB%8F%84%ED%99%94-%EC%9E%91%EC%97%85
Java에서는 스레드 로컬(ThreadLocal)을 이용하여 구현합니다. 이전 포스팅에서 잠깐 언급되어 좀 더 부가적으로 설명을 드리자면 스레드 로컬이란 해당 스레드만 접근할 수 있는 저장소로, 각 스레드마다 별도의 내부 저장소가 제공되어 여러 스레드가 같은 객체에 접근했을 때 데이터의 유실 문제를 해결할 수 있습니다. 따라서 이전 포스팅의 SecurityContextHolder의 경우 스레드 로컬 방식으로 여러 SecurityContext가 저장되어도 하나의 SecurityContextHolder 객체 내에서 데이터가 유실되지 않고 각각을 저장할 수 있게 됩니다.

Spring Boot에서는 이를 MDC(Mapped Diagnostic Context)를 이용하여 접근할 수 있습니다. MDC는 map 형식으로 여러 데이터를 넣을 수 있어 요청 URI, 요청 IP 등 로깅에 필요한 정보를 유동적으로 저장하여 로그 추적을 더 쉽게 할 수 있습니다.

그럼 MDC를 이용하여 Slack 에러 로그 전송 시스템을 수정해보도록 하겠습니다. 우선 ThreadPoolTaskExecutor에서 MDC를 정상적으로 읽을 수 있도록 TaskDecorator를 implements한 CustomTaskDecorator 클래스를 생성하겠습니다. 위에서 살펴본 것처럼 MDC는 스레드 로컬 방식으로 동작하기 때문에 같은 스레드 내에서만 참조가 가능합니다. 따라서 ThreadPoolTaskExecutor를 이용할 경우 스레드가 변경되어 정상적으로 MDC를 참조할 수 없어 기존 스레드의 MDC값을 ThreadPoolTaskExecutor로 복사하는 과정이 필요합니다.

import org.slf4j.MDC;
import org.springframework.core.task.TaskDecorator;

import java.util.Map;

public class CustomTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable task) {
        Map<String, String> callerThreadContext = MDC.getCopyOfContextMap();
        return () -> {
            if(callerThreadContext != null) MDC.setContextMap(callerThreadContext);
            task.run();
        };
    }
}

위에서 생성한 CustomTaskDecorator 클래스를 ThreadPoolConfig에 추가합니다. 여기서 @EnableAsync를 이용하여 이전 포스팅에서처럼 ThreadPoolTaskExecutor 객체를 직접 생성하지 않고 @Async 어노테이션을 붙임으로서 해당 메소드를 ThreadPoolTaskExecutor로 실행시킬 수 있습니다.

@Configuration
@EnableAsync
public class ThreadPoolConfig {
    @Value("${spring.task.execution.pool.core-size}")
    int CORE_POOL_SIZE;
    @Value("${spring.task.execution.pool.max-size}")
    int MAX_POOL_SIZE;
    @Value("${spring.task.execution.pool.queue-capacity}")
    int QUEUE_CAPACITY;

    @Bean
    public ThreadPoolTaskExecutor taskExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setQueueCapacity(QUEUE_CAPACITY);

        executor.setTaskDecorator(new CustomTaskDecorator());
        executor.setThreadNamePrefix("async-task-");
        executor.setThreadGroupName("async-group");

        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.initialize();
        return executor;
    }
}

아래는 MDC에 추가할 정보들입니다. 여기서 유의할 점은 MDC에 저장할 때 키로 설정된 값에 ID가 필요없다는 것입니다. 위에서 살펴본 것처럼 스레드 로컬 특성상 각 스레드마다 별도의 내부 저장소가 제공되어 같은 객체에 접근해도 독립적으로 참조하기 때문입니다. Spring Boot 특성 상 (기본적으로) Request 하나 당 스레드 하나가 사용되기 때문에 자연스럽게 요청마다 모두 MDC값이 다르게 됩니다.

    public void setMdc(HttpServletRequest request){
        MDC.put("request_id", UUID.randomUUID().toString());
        logger.info("request_id : " + MDC.get("request_id"));

        MDC.put("request_context_path", request.getContextPath());
        logger.info("request_context_path : " + MDC.get("request_context_path"));

        MDC.put("request_url", request.getRequestURI());
        logger.info("request_url : " + MDC.get("request_url"));

        MDC.put("request_method", request.getMethod());
        logger.info("request_method : " + MDC.get("request_method"));

        MDC.put("request_time", new Date().toString());
        logger.info("request_time : " + MDC.get("request_time"));

        MDC.put("request_ip", request.getRemoteAddr());
        logger.info("request_ip : " + MDC.get("request_ip"));

        MDC.put("request_header", request.getHeader(TokenProvider.HEADER_NAME));
        logger.info("request_header : " + MDC.get("request_header"));

        MDC.put("request_query_string", request.getQueryString());
        logger.info("request_query_string : " + MDC.get("request_query_string"));
    }

아래는 Slack에서 메시지를 전송하는 코드입니다. @Async를 통해 Configuration을 진행한 ThreadPoolTaskExecutor로 실행되어 비동기로 처리됩니다. 이 때 기존 request의 context가 넘어가지 않고 새로 만들어지기 때문에 MDC에 저장한 값을 사용합니다.

    @Async
    public void sendSlack(Exception e) {
        slackAttachment.setTitleLink(MDC.get("request_context_path"));
        slackAttachment.setText(Arrays.toString(e.getStackTrace()));

        slackAttachment.setFields(
                List.of(
                        new SlackField().setTitle("Request URL").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request Method").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request Time").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request IP").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request header").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request Query String").setValue(MDC.get("request_context_path")),
                        new SlackField().setTitle("Request Body").setValue(MDC.get("request_context_path"))
                )
        );

        slackMessage.setAttachments(Collections.singletonList(slackAttachment));
        slackApi.call(slackMessage);
        mdcProvider.removeMdc();
    }

아래는 Slack에 전달된 메시지입니다. 성공적으로 파라미터들이 등록된 것을 확인할 수 있습니다. (테스트에 사용된 API의 경우 헤더, 쿼리스트링, 바디가 없어 빈 값으로 표시됩니다.)

주의할 점은 Test 클래스에서 MockMvc를 이용하여 ThreadPoolTaskExecutor 내의 Slack 메시지 전송 메소드를 실행 시 전송이 되지 않는 경우를 발견했습니다. 명확한 원인은 모르겠으나 Test 외에 서비스 실행 시 정상적으로 작동하는 것을 확인하였고 Test 메소드 내에서 ThreadPoolTaskExecutor 내와 현재 스레드 총 2번을 연속해서 메시지를 보낼 경우 2개 모두 수신되었습니다. 또한 어떤 경우는 ThreadPoolTaskExecutor만 설정하여도 메시지가 수신되는 경우가 있어 이 부분은 추후 더 정확한 원인을 찾아보도록 하겠습니다.

그럼 Jmeter를 이용한 접속 테스트를 진행해보겠습니다.(Jmeter 사용법은 이전 포스팅을 참고해주세요 : https://velog.io/@gale4739/Spring-Boot-Jmeter-%EC%82%AC%EC%9A%A9M1) 실험 조건은 10개의 스레드(요청 수)를 2번씩 반복하여 총 20개의 request가 도달하도록 진행하였으며. 각 요청은 Slack에 오류 메시지를 전송하고 MDC의 UUID값을 반환합니다.

아래의 캡쳐본은 테스트를 돌린 결과입니다. 보시는 것처럼 같은 유저가 전송한 요청의 UUID값이 서로 다른 것을 확인할 수 있습니다. 이를 통해 MDC가 서로 영향을 받지 않는 독립적으로 작동한다는 것을 확인하였으며 할당된 범위 내에서 정상적으로 작동한다는 것을 확인할 수 있습니다.

스레드 로컬 및 스레드 풀에 대해 더 알아보며 스레드 로컬이 안정적으로 동작할 수 있는 limit을 컨트롤해 더 정확한 트래픽 처리를 공부해야 할 것 같습니다. 그럼 이상으로 오늘의 포스팅 마치도록 하겠습니다!

참고 자료
https://bcho.tistory.com/1316
https://blog.gangnamunni.com/post/mdc-context-task-decorator/
https://velog.io/@dbsrud11/SpringBoot-%ED%95%B5%EC%8B%AC-%EC%9B%90%EB%A6%AC-%EC%93%B0%EB%A0%88%EB%93%9C-%EB%A1%9C%EC%BB%AC-ThreadLocal-2