AbstractAggregateRoot와 ApplicationEventPublisher 동작 과정

개요

스프링의 이벤트 기능을 활용할 때 흔히 AbstractAggregateRoot 혹은 ApplicationEventPublisher를 사용한다. 이 글에서는 내부적으로 어떤 방식으로 동작하며, 이벤트가 어떻게 등록되고 발행되어 최종적으로 수신되는지 전체적인 과정을 정리해 본다.

 

AbstractAggregateRoot 동작 과정

AbstractAggregateRoot는 Spring Data에서 제공하는 추상 클래스로, 도메인 이벤트를 보다 손쉽게 등록하고 관리할 수 있도록 도와준다. 주로 DDD에서 상태 변경 후 외부에 의미 있는 이벤트를 전달할 때 사용된다. 

예를 들어 주문이 완료되었을 때 도메인 이벤트를 등록하는 코드는 다음과 같다.

@Entity
public class Order extends AbstractAggregateRoot<Order> {
    public void complete() {
        this.status = OrderStatus.COMPLETED;
        // 도메인 이벤트 등록
        registerEvent(new OrderCompletedEvent(this.getId(), /*...*/));
    }
}

위 코드처럼 AbstractAggregateRoot.java를 상속한 엔티티는 registerEvent() 메서드를 통해 도메인 이벤트를 내부적으로 관리되는 리스트(domainEvents)에 추가할 수 있다. 

https://github.com/spring-projects/spring-data-commons/blob/main/src/main/java/org/springframework/data/domain/AbstractAggregateRoot.java#L48

그리고 이렇게 등록된 이벤트는 다음과 같이 Spring Data의 save(), saveAll(), delete(), deleteAll(), deleteAllInBatch(), deleteInBatch() 등의 메서드가 호출될 때 자동으로 추출되어 발행된다.(참고로 deleteById()의 경우에는 호출되지 않는다. 이유는 #issues/2084를 확인하자.)

@Transactional
public void completeOrder(Order order) {
    order.complete();          // 상태 변경 + registerEvent(...)
    orderRepository.save(order);   // save() 호출 → 이벤트 발행
}

@Transactional
public void cancelOrder(Order order) {
    order.cancel();            // cancel() 내부에서 이벤트 등록
    orderRepository.delete(order); // delete() 호출 → 이벤트 발행
}

https://docs.spring.io/spring-data/jpa/reference/repositories/core-domain-events.html

이건 개인적인 생각인데 사용하면서 아쉬운 점은 도메인 내부에서 updateXxx()와 같은 메서드로 이벤트를 등록하더라도 save()나 delete()를 명시적으로 호출하지 않으면 이벤트가 발행되지 않는다는 점이다. JPA의 더티 체킹을 통해 상태 변경은 반영되더라도, 업데이트 행위에 불필요한 save()나 delete() 호출을 추가해야 하는 경우가 생겨 구조적으로 다소 어색하게 느껴질 수 있다.

여튼 이후 도메인 이벤트가 실제로 발행되기 위해서는, 엔티티의 저장 또는 삭제 시점에 Spring Data 내부에서 이벤트를 추출하고 이를 ApplicationEventPublisher를 통해 발행하는 과정이 필요하다. 그리고 이 전체 흐름을 담당하는 핵심 클래스가 바로 EventPublishingRepositoryProxyPostProcessor.java이다.

https://github.com/spring-projects/spring-data-commons/blob/main/src/main/java/org/springframework/data/repository/core/support/EventPublishingRepositoryProxyPostProcessor.java#L72

Spring Data는 사용자 정의 리포지토리를 내부적으로 프록시 객체로 감싸는데, 이 프록시 생성 과정에서 EventPublishingRepositoryProxyPostProcessor가 개입하여 이벤트 처리를 위한 인터셉터를 등록한다. 구체적으로는 위의 postProcess() 메서드가 호출되며, 이 시점에 EventPublishingMethodInterceptor가 프록시에 추가된다.

이후 실제 애플리케이션에서 repository.save(entity) 또는 repository.delete(entity)가 호출되면, 해당 인터셉터의 invoke() 메서드가 실행된다.

https://github.com/spring-projects/spring-data-commons/blob/main/src/main/java/org/springframework/data/repository/core/support/EventPublishingRepositoryProxyPostProcessor.java#L107

이 invoke() 메서드는 먼저 실제 DB 저장 로직을 수행한 뒤, 대상 엔티티에 @DomainEvents 애노테이션이 붙은 메서드나 AbstractAggregateRoot의 domainEvents() 메서드를 호출하여 이벤트 목록을 추출한다. 이벤트가 존재할 경우 하나씩 ApplicationEventPublisher.publishEvent()를 통해 발행하고, 마지막으로 @AfterDomainEventPublication이 붙은 메서드를 호출해 내부 이벤트 리스트를 초기화한다.

결국 도메인 이벤트가 실제로 발행되는 시점은 아래 EventPublishingMethod.publishEventsFrom() 메서드에서 ApplicationEventPublisher.publishEvent(...)가 호출될 때다.

https://github.com/spring-projects/spring-data-commons/blob/main/src/main/java/org/springframework/data/repository/core/support/EventPublishingRepositoryProxyPostProcessor.java#L195

이러한 흐름을 통해 도메인 계층에서는 registerEvent()로 이벤트만 등록하면 되고, save()나 delete() 시점에 Spring Data가 이를 자동으로 추출·발행하는 구조가 완성된다.

위 내용들을 한줄로 요약하면 AbstractAggregateRoot에서 registerEvent()로 등록된 이벤트는, save() 또는 delete() 메서드가 호출된 직후 트랜잭션 안에서 즉시 Spring Data JPA가 ApplicationEventPublisher를 통해 발행된다.

 

ApplicationEventPublisher 동작 과정

ApplicationEventPublisher.publishEvent(event)가 호출되는 순간부터, 이벤트는 스프링 컨테이너 내부 파이프라인을 따라 전파된다. 전체적인 흐름은 다음과 같다.

① ApplicationContext.publishEvent()

ApplicationContext는 ApplicationEventPublisher를 구현한다. 사용자가 publishEvent(event)를 호출하면 컨텍스트가 가장 먼저 이벤트를 받아들이는 진입점이 되며, 실제 처리는 내부적으로 AbstractApplicationContext.java의 publishEvent()를 통해 처리된다.

② AbstractApplicationContext.publishEvent()

전달된 객체가 ApplicationEvent를 상속하지 않은 순수 POJO라면, 이를 PayloadApplicationEvent로 래핑 한다. 그 이유는 Spring 4.2 이전에는 이벤트를 발행하려면 반드시 ApplicationEvent 클래스를 상속해야 했으나 v4.2부터는 단순한 객체도 이벤트로 발행할 수 있게 되었다. 

// Spring 4.2 이전: ApplicationEvent 상속 필요
public class OrderCompletedEvent extends ApplicationEvent {
    private final Long orderId;
    public OrderCompletedEvent(Object source, Long orderId) {
        super(source);
        this.orderId = orderId;
    }
}

// 4.2 이후: 아무 POJO나 바로 발행 가능
record OrderCompleted(Long orderId) {}
publisher.publishEvent(new OrderCompleted(1L));

다만 내부 이벤트 전달 체계는 여전히 ApplicationEvent 기반으로 동작하기 때문에, POJO를 직접 사용할 수 있게 하면서도 기존 메커니즘과 호환성을 유지하려면 중간에 PayloadApplicationEvent라는 래핑이 필요하다. 이러한 이유로 스프링은 ApplicationEvent가 아닌 일반 객체가 전달되면 이를 PayloadApplicationEvent로 감싸 내부 이벤트 처리 흐름에 맞춰 전파한다.

③, ④   SimpleApplicationEventMulticaster.multicastEvent()

이벤트가 ApplicationEventMulticaster에 도달하면, 기본 구현체인 SimpleApplicationEventMulticaster가 사용되고, 크게 다음과 같이 두 가지 동작이 일어난다.

https://github.com/spring-projects/spring-framework/blob/main/spring-context/src/main/java/org/springframework/context/event/SimpleApplicationEventMulticaster.java#L137

1. 리스너 목록 결정 – getApplicationListeners(event, type)로 현재 이벤트에 반응할 리스너를 필터링한다.
2. 실행 방식 결정 – for-loop를 돌며 각 리스너를 동기로 호출할지 스레드 풀에서 비동기로 호출할지 판단한다.

그리고 비동기의 전환은 아래 두 가지 조건을 만족했을 때 이루어진다.

첫 번째는 TaskExecutor가 설정되어 있는지 여부이다. 컨텍스트 초기화 시 이름이 applicationEventMulticaster 인 빈을 직접 등록하고 setTaskExecutor(...)로 스레드 풀을 주입하면, 멀티캐스터는 스레드 풀을 사용하여 비동기로 처리할 수 있게 된다. 빈을 따로 정의하지 않으면 기본 SimpleApplicationEventMulticaster가 생성되는데, 이때 taskExecutor == null이므로 모든 리스너는 동기로 실행된다.

@Configuration
public class EventConfig {

    // 스레드 풀 정의
    @Bean
    public ThreadPoolTaskExecutor eventPool() {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        pool.setCorePoolSize(4);
        pool.setMaxPoolSize(8);
        pool.initialize();
        return pool;
    }

    // 전역 이벤트 멀티캐스터에 스레드 풀 주입
    @Bean(name = "applicationEventMulticaster")
    public ApplicationEventMulticaster multicaster(ThreadPoolTaskExecutor eventPool) {
        var m = new SimpleApplicationEventMulticaster();
        m.setTaskExecutor(eventPool); // 비동기 실행 가능
        return m;
    }
}

위처럼 설정한 경우, 멀티캐스터는 TaskExecutor가 있으므로 비동기 실행이 가능하다고 판단한다. 만약 위와 같이 별도 빈 등록 없이 기본 설정을 사용할 경우에는 taskExecutor == null 상태로 등록되어 항상 동기로 실행된다.(다만 나는 지금까지 프로젝트를 진행하면서 위와 같이 applicationEventMulticaster를 직접 정의한 경우는 보지 못했다. 보통 Listener 측에 @Async를 통해 비동기로 수행했다.)

두 번째는 각 리스너가 supportsAsyncExecution()에서 true를 반환하는지 여부이다. 일반적인 @EventListener 기반 리스너는 기본적으로 true를 반환하기 때문에 비동기 처리가 가능하지만, @TransactionalEventListener의 경우에는 트랜잭션 보장을 위해 실제로 내부 구현에서는 supportsAsyncExecution()이 false로 고정되어 false를 반환한다. 다음 코드들을 살펴보자.

@Component
public class MailNotifier {

    @EventListener
    public void handle(OrderCompletedEvent e) {
        // 비동기 실행 가능 (전역 풀이 있을 경우)
    }
}

만약 applicationEventMulticaster 빈에 스레드 풀을 주입해 주었다면 위 메서드는 비동기로 실행될 것이고, 그렇지 않다면 호출 스레드에서 동기로 실행된다.

@EnableAsync
@Component
public class SmsNotifier {

    @Async
    @EventListener
    public void sendSms(OrderCompletedEvent e) {
        // 멀티캐스터는 동기 호출하지만, @Async 덕분에 메서드 본문만 별도 풀에서 비동기로 처리
    }

    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void sendSms(OrderCompletedEvent e) {
        // supportsAsyncExecution() == false 이므로 호출 자체는 호출 스레드
        // 그러나 @Async 로 인해 별도 풀에서 실행
    }
    
}

보통은 이렇게 @Async 어노테이션을 사용해서 비동기로 작업을 할 것이다. 이 경우 멀티캐스터는 동기적으로 호출하지만, @Async 덕분에 본문은 별도 스레드 풀에서 비동기 처리되며, 전역 TaskExecutor 설정과는 무관하게 비동기화가 가능하다.

⑤ SimpleApplicationEventMulticaster.invokeListener()

이 단계는 리스너 메서드를 정말로 호출하는 마지막 진입점이다. 이 메서드에서 다음과 같은 동작이 수행된다.

https://github.com/spring-projects/spring-framework/blob/main/spring-context/src/main/java/org/springframework/context/event/SimpleApplicationEventMulticaster.java#L162

• ErrorHandler가 없음(기본) : 동기 실행일 경우 예외가 발생한 리스너에서 즉시 상위로 전파되고, 이후 리스너는 실행되지 않는다. 비동기 실행일 경우에는 각 리스너가 별도 스레드에서 실행되므로, 한 리스너의 예외가 다른 리스너 실행에 영향을 주지 않는다.
• ErrorHandler가 있음 : 예외를 핸들러로 위임하여 로깅하거나 무시할 수 있으며, 다른 리스너들의 실행이 계속 진행된다.

https://github.com/spring-projects/spring-framework/blob/main/spring-context/src/main/java/org/springframework/context/event/SimpleApplicationEventMulticaster.java#L178

이후 doInvokeListener(listener, event)를 통해 실제로 ApplicationListener.onApplicationEvent()가 호출된다. 이 메서드 안에서는 리스너 타입과 이벤트 타입이 일치하지 않아 ClassCastException이 발생할 수 있는데,  타입 불일치일 경우 TRACE 로그만 남기고 조용히 예외를 무시한다. 그렇지 않은 경우에는 예외를 다시 던져 호출자에게 알린다.

또한 이벤트가 비동기로 처리되도록 설정되었더라도, 스레드 풀이 가득 차거나 종료되어 RejectedExecutionException이 발생한 경우, 스프링은 이 예외를 그대로 던지지 않고 현재 스레드에서 동기적으로 이벤트를 실행하는 방식으로 우회 처리한다. 이로써 애플리케이션 종료 시점 등에서도 이벤트 손실 없이 안전한 처리를 보장할 수 있다.

⑥ 리스너 수행

멀티캐스터는 최종적으로 어댑터를 통해 해당 @EventListener 또는 @TransactionalEventListener 메서드를 실행한다. 참고로 @TransactionalEventListener 같은 경우는 phase 속성에 따라 이벤트 리스너의 실행 시점을 지연시킬 수 있다.

그럼 여기서 "@TransactionalEventListener는 어떻게 멀티캐스터 단계에서 바로 실행되지 않고 phase 설정에 따라 실행 시점을 미룰 수 있을까?" 하는 궁금증이 들 수 있을텐데, 이 질문에 대한 정답은 @TransactionalEventListener를 감싸는 내부 어댑터 클래스인 TransactionalApplicationListenerMethodAdapter.java에 있다.

SimpleApplicationEventMulticaster는 모든 리스너에 대해 단순히 onApplicationEvent() 메서드를 호출할 뿐이다. 그런데 @TransactionalEventListener의 경우는 이 어댑터가 해당 인터페이스를 구현하고 있어서 이벤트가 발생해도 리스너 메서드를 즉시 실행하지 않고, 트랜잭션에 후속 동작으로 등록만 해두는 방식으로 처리된다. 등록 후에  SimpleApplicationEventMulticaster의 역할은 끝이난다.

https://github.com/spring-projects/spring-framework/blob/main/spring-tx/src/main/java/org/springframework/transaction/event/TransactionalApplicationListenerMethodAdapter.java#L85

더 자세히 설명하면 해당 본문의 주제를 벗어나는 것 같아 간략하게만 살펴보면 AFTER_COMMIT로 등록된 리스너는 이벤트가 발행되더라도 즉시 실행되지 않고, TransactionSynchronizationManager를 통해 해당 리스너를 트랜잭션 후속 작업으로 사용할 수 있는 콜백 객체(TransactionalApplicationListenerSynchronization)로 등록해 둔다. 그리고 트랜잭션이 커밋되는 시점이 되면 PlatformTransactionManager는 등록된 콜백들의 afterCommit() 메서드를 호출하고,
그 안에서 listener.processEvent(event) → invokeListenerMethod()를 통해 최종적으로 리스너 메서드가 실행된다.