Spring @EventListener 사용하기 (2)

 

https://jinmook.tistory.com/29

Spring @EventListener 사용하기

 

 

앞선 글에서는 이벤트 리스너와 비동기 설정, 트랜잭션 설정 등에 관해 알아봤습니다.
이번 글에서는 이어서 트랜잭션과 관련된 내용 그리고 Tomcat과 Executor 관계에 대해 알아보겠습니다.

 

 

글 목차

1. @TransactionalEventListener

2. TaskExecutor

3. @Async를 이용한 Executor 지정

4. 전체 tomcat 스레드와 Executor 관계

 

 

 

---

 

@TransactionalEventListener

 

가장 먼저 @TransactionalEventListener 어노테이션을 사용하면 해당 어노테이션 내부에 @EventListener 어노테이션이 존재하기 때문에 추가로 @EventListener 어노테이션을 작성하지 않아도 됩니다.

 

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})  
@Retention(RetentionPolicy.RUNTIME)  
@Documented  
@EventListener  
public @interface TransactionalEventListener {
    ...
}

 

 

@TransactionalEventListener 어노테이션의 phase 설정을 통해 기존 트랜잭션의 상태에 따른 Listener 실행 시점을 조절할 수 있습니다. 크게 4가지 실행 시점이 있습니다. 이때 기본값은 AFTER_COMMIT 입니다.

• BEFORE_COMMIT
• AFTER_COMMIT
• AFTER_ROLLBACK
• AFTER_COMPLETION

 

이 중에서 AFTER_COMPLETION 의 시점은 AFTER_COMMIT, AFTER_ROLLBACK 각각의 시점을 합친거라고 생각할 수 있습니다.

 

 

 

동일 스레드에서의 @TransactionalEventListener

이제 LIstener 부분에서 트랜잭션이 어떻게 실행되는지 확인해보겠습니다. 우선 Listener 부분에서도 DB 관련 로직이 필요하다는 가정하에 아래 코드를 살펴보겠습니다.

 

@Slf4j  
@Component  
@RequiredArgsConstructor  
public class HelloEventListener {  

    private final ListenerRepository listenerRepository;  

    @TransactionalEventListener
    public void listen(HelloEventDto helloEventDto) throws InterruptedException {  

        String transactionName = TransactionSynchronizationManager.getCurrentTransactionName();  
        log.info("HelloEventListener transactionName = {}", transactionName);  

        log.info("eventListener start");  
        ListenerEntity listenerEntity = ListenerEntity.from(helloEventDto);  

        Thread.sleep(5000);  

        ListenerEntity savedListenerEntity = listenerRepository.save(listenerEntity);  
        log.info("savedListenerEntity = {}", savedListenerEntity);  
    }  
}

 

 

아래 로그를 살펴보면 ListenerEntity 에 대한 insert 쿼리가 없는것을 확인할 수 있습니다.

 

 

 

이미 이전 트랜잭션이 commit 을 한 이후에 Listener 로직이 실행되기 때문에 해당 entity는 저장되지 않는다는 것을 확인할 수 있었습니다.

 

 

따라서 LIstener 로직에서도 DB 영속성을 사용하고 싶다면 @Transactional 어노테이션의 propagation 설정을 REQUIRES_NEW 로 설정하여 새로운 트랜잭션으로 이용해야 합니다. propagation 설정을 기본 설정으로 진행하면 @TransactionalEventListener 어노테이션과 같이 사용할 수 없다는 에러가 발생하며 어플리케이션이 실행되지 않습니다.

 

 

그럼 만약에 @TransactionalEventLIstener(phase = TransactionPhase.BEFORE_COMMIT) 으로 설정하면 어떻게 될까??

 

 

 

위 사진을 통해 알 수 있듯이 해당 LIstener 관련 로직이 커밋 전에 진행되기 때문에 Listener 에서 작성된 영속성 저장 로직도 정상적으로 커밋되는 것을 확인할 수 있습니다.

 

 

또한 LIstener 로직이나 원래 진행하던 메인 서비스 로직에서 에러가 발생하게 되면 전체 로직이 롤백이 되게 됩니다. 어찌보면 하나의 트랜잭션에 포함되었기 때문에 스프링에서 @Transactional 어노테이션의 작동 방식을 생각하면 자연스럽게 생각할 수 있는 결과였습니다.

 

 

지금까지는 비동기가 아닌 동일 스레드에서 LIstener 로직이 작동하는 방식에서 @TransactionalEventListener 어노테이션의 동작 방식을 살펴봤습니다. 다음으로는 비동기 상태에서의 @TransactionalEventLIstener 어노테이션의 동작 방식에 대해 알아보겠습니다.

 

 

다른 스레드에서의 @TransactionalEventListener

LIstener 부분 코드는 아래와 같습니다. @Async 어노테이션과 @Transactional 어노테이션을 추가했습니다. 이때 @Transactional 어노테이션의 propagation = Propagation.REQUIRES_NEW 설정으로 진행합니다.

 

@Async  
@Transactional(propagation = Propagation.REQUIRES_NEW)  
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)  
public void listen(HelloEventDto helloEventDto) throws InterruptedException {  

    String transactionName = TransactionSynchronizationManager.getCurrentTransactionName();  
    log.info("HelloEventListener transactionName = {}", transactionName);  

    log.info("eventListener start");  
    ListenerEntity listenerEntity = ListenerEntity.from(helloEventDto);  

    Thread.sleep(5000);  

    ListenerEntity savedListenerEntity = listenerRepository.save(listenerEntity);  
    log.info("savedListenerEntity = {}", savedListenerEntity);  

    throw new RuntimeException("리스터 에러 발생");  
}

 

 

위 코드를 실행하면 Listener 로직에서 에러가 발생합니다. 이때 메인 로직의 트랜잭션과 Listener 로직의 트랜잭션이 서로 다른 트랜잭션이기 때문에 메인 로직의 트랜잭션은 정상적으로 커밋하고 LIstener 로직의 트랜잭션은 롤백되는 것을 확인할 수 있습니다.

 

 

지금까지 @TransactionalEventLIstener 어노테이션과 트랜잭션과의 관계에 대해 알아봤습니다.

다음으로는 스레드와의 관계의 대해 알아보도록 하겠습니다.

 

 

---

 

TaskExecutor

 

가장 먼저 Spring 공식 문서의 내용을 간단하게 살펴보겠습니다.
TaskExecutor

Task Execution and Scheduling :: Spring Framework

 

 

스프링에서는 TaskExecutor 그리고 TaskScheduler 인터페이스를 통해 비동기 작업 실행에 대한 추상화를 제공하고 있습니다. Executors 는 스레드 풀 개념에 대한 JDK 명칭입니다. Executor는 단일 스레드일 수도 있고 심지어 동기적일 수도 있습니다.

 

 

TaskExecutor 는 원래 Spring의 다른 구성 요소에 스레드 풀링이 필요한 경우 이를 추상화하기 위해 만들어졌습니다. ApplicationEventMulticaster, JMS의 AbstractMessageListenerContainer, Quartz 통합과 같은 구성 요소는 모두 스레드 풀을 위해 TaskExecutor 추상화를 사용합니다.

 

 

해당 문서에서 여러 TaskExecutor 구현체들에 대한 설명이 나옵니다. 간단하게 하나씩 살펴보겠습니다.

 

• SyncTaskExecutor : 이 구현체는 호출을 비동기적으로 실행하지 않습니다. 대신 각 호출이 호출하는 스레드에서 실행됩니다. 주로 멀티스레딩이 필요하지 않은 간단한 테스트 케이스에서 사용됩니다.
• SimpleAsyncTaskExecutor : 이 구현체는 스레드를 재사용하지 않습니다. 대신 각 호출마다 새로운 스레드를 시작합니다. 하지만 동시성 제한을 지원하여, 제한을 초과하는 호출은 슬롯이 비워질 때까지 차단됩니다. 진정한 폴링을 원한다면 ThreadPoolTaskExecutor를 참조해야 합니다.
• ConcurrentTaskExecutor : 이 구현체는 java.util.concurrent.Executor 인스턴스의 어댑터입니다. ConcurrentTaskExecutor를 직접 사용할 필요는 거의 없습니다. 하지만 ThreadPoolTaskExecutor가 충분히 유연하지 않다면, ConcurrentTaskExecutor가 대안이 될 수 있습니다.
• ThreadPoolTaskExecutor : 가장 일반적으로 사용되는 구현체입니다. 참고로 Spring의 라이프사이클 관리 기능을 통해 일시 정지/재개 기능과 우아한 종료 기능을 제공합니다.
• DefaultManagedTaskExecutor : JSR-236 호환 런타임 환경에서 JNDI로 얻은 ManagedE#xecutorService를 사용하여 CommonJ WorkManager 대체합니다.

 

참고로 SimpleAsyncTaskExecutor 의 경우 JDK 21의 가상 스레드와 맞춘 virtualThreads 옵션을 제공하며, 우아한 종료 기능도 제공합니다.

 

 

---

 

@Async를 이용한 Executor 지정

 

기본적으로 Spring 에서는 다른 설정이 없다면 TaskExecutorConfigurations 클래스를 통해 TaskExecutor를 자동 주입하고 있습니다. 직접 Executor를 생성해서 빈으로 등록하고 싶다면, 설정 클래스에서 빈을 생성하거나 혹은 AsyncConfigurer 인터페이스를 사용하여 원하는 Executor 를 빈으로 등록하여 사용할 수 있습니다. 여기서는 AsyncConfigurer 인터페이스를 사용하는 예시를 설명하겠습니다.

 

 

아래 코드는 AsyncConfigurer 인터페이스를 구현한 설정 클래스입니다.

 

@Configuration  
@EnableAsync  
public class AsyncConfig implements AsyncConfigurer {  

    @Override  
    public Executor getAsyncExecutor() {  
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();  
        executor.setCorePoolSize(5); // thread-pool에 항상 살아있는 thread 최소 개수  
        executor.setMaxPoolSize(5); // thread-pool에서 사용 가능한 최대 thread 개수  
        executor.setQueueCapacity(500); // thread-pool에서 사용할 최대 queue 크기  
        executor.setThreadNamePrefix("custom-task-");  
        executor.initialize();  
        return executor;  
    }  
}

 

 

위 설정을 작성하고 테스트를 진행하면 스레드 이름이 custom-task-1 형식으로 만들어 지는 것을 아래 로그 사진을 통해 확인할 수 있습니다.

 

 

 

위 설정을 통해 @Async 어노테이션을 사용하면 위에서 등록한 Executor를 사용하게 됩니다. 이는 어플리케이션 전역으로 스레드를 설정하는데 메서드 단위로도 스레드를 설정할 수 있습니다. @Async 의 value 속성을 이용해 사용하고자 하는 Executor 빈 이름을 작성해주면 됩니다.

 

 

위 설정 코드에 아래 처럼 추가로 빈 등록을 진행합니다. 이후 기존 Listener 코드에 존재하는 @Async 어노테이션에 설정값을 추가합니다.

 

// config 파일에 추가
@Bean  
public Executor customExecutor() {  
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();  
    executor.setCorePoolSize(5); // thread-pool에 항상 살아있는 thread 최소 개수  
    executor.setMaxPoolSize(5); // thread-pool에서 사용 가능한 최대 thread 개수  
    executor.setQueueCapacity(500); // thread-pool에서 사용할 최대 queue 크기  
    executor.setThreadNamePrefix("method-task");  
    executor.initialize();  
    return executor;  
}

// Listener 코드에 @Async 어노테이션 설정에 위에서 정의한 cutomExecutor 빈 이름을 작성
@Async("customExecutor")  
@EventListener  
public void listen(HelloEventDto helloEventDto) throws InterruptedException {  
    // 리스너 로직 진행
}

 

 

위 코드로 설정을 진행하게되면 추가로 빈으로 등록한 customExecutor 가 사용되게 됩니다. 따라서 로그에서도 위에서 정의한 prefix인 method-task prefix가 붙은 형태로 로그가 찍히게 됩니다. 아래 사진을 통해 확인 가능합니다.

 

 

 

---

 

전체 tomcat 스레드와 Executor 스레드 관계

위에서 잠깐 언급했지만 tomcat 스레드와 Executor 스레드는 전혀 관계가 없습니다. Tomcat 은 요청을 처리하기 위해 내부적으로 스레드풀을 사용하고 있으며, 이를 통해 여러 요청을 동시에 처리할 수 있습니다.

 

Executor 스레드풀의 경우 Spring Boot에서 @Async 어노테이션을 사용하거나 스레드를 명시적으로 생성하여 비동기 작업을 처리할 때 Executor를 사용할 수 있습니다. 이 Executor는 Tomcat의 스레드풀과는 별개의 스레드풀을 사용하며, 주로 애플리케이션 내부의 비동기 작업을 처리하는 데 사용합니다.

 

 

따라서 두 스레드에 대해 정리해보자면 아래와 같습니다.

• Tomcat 스레드풀 : HTTP 요청을 처리하는 스레드풀입니다. maxThreads, minSpareThreads, acceptCount 등의 설정을 진행할 수 있습니다.
  • 참고로 TomcatWebServerFactoryCustomizer 클래스를 통해 스레드풀을 생성하는 것을 확인할 수 있습니다.

 

• Executor 스레드풀 : 애플리케이션 내부의 비동기 작업을 처리하는 스레드풀입니다. @Async 어노테이션과 함께 사용됩니다.

 

결과적으로 두 스레드풀은 독립적이지만, 애플리케이션의 성능 최적화를 위해 함께 조정될 수 있습니다.