Spring Events를 사용해 이벤트 발행하기

택시비 절약을 위해 함께 택시를 탑승할 학생들을 모집하는 '택시팟' 서비스를 개발하면서 Spring Events를 사용했던 경험에 대해 정리해보려고 합니다. 해당 프로젝트 코드는 GitHub에서 확인할 수 있습니다.

Spring Events

스프링 이벤트는 애플리케이션 내부에서 이벤트를 발행해 다른 객체로 데이터를 전달할 수 있는 기능입니다. ApplicationContext에서 제공하는 기능으로 ApplicationEventPublisher를 주입받아 간단히 사용할 수 있습니다. 왠지 모르겠지만, 공식문서에서 스프링 이벤트와 관련해 자세히 설명된 부분을 찾지는 못 하겠더라구요. 하지만 사용법은 간단하기 때문에 몇 가지 주의사항을 제외하면 유용하게 활용할 수 있었습니다.

ApplicationEventPublisher 인터페이스를 살펴보면 매개변수 타입이 다른 두 개의 publishEvent() 메서드가 있습니다.

@FunctionalInterface
public interface ApplicationEventPublisher {
	
    default void publishEvent(ApplicationEvent event) {
		publishEvent((Object) event);
	}
    
    void publishEvent(Object event);
}

Object 타입 매개변수를 가진 메서드는 Spring Framework 4.2 버전부터 지원되기 때문에 4.2 이전 버전(4.1 ≥)에서는 이벤트 객체가 반드시 ApplicationEvent 추상 클래스를 상속하도록 구현해야 합니다.

간단히 아래와 같이 이벤트를 발행할 수 있습니다.

@Component
public class ExampleEventPublisher {

    private final ApplicationEventPublisher eventPublisher;

    public ExampleEventPublisher(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    public void publishExampleEvent() {
        ExampleEvent event = new ExampleEvent();
        eventPublisher.publishEvent(event); // 이벤트 발행
    }
}

또한, 4.2 이전 버전에서는 이벤트 리스너를 ApplicationListener 인터페이스를 구현하도록 작성해야 했지만 4.2 버전부터는 @EventListener 어노테이션을 사용해 이벤트 핸들링을 할 수 있습니다. 처리할 이벤트의 타입은 매개변수를 통해 자동으로 감지합니다!

@Component
public class ExampleEventSubscriber {

    @EventListener
    public void handleExampleEvent(ExampleEvent event) {
        // 이벤트 처리
    }
}

사용 배경

택시팟 서비스에서 사용자가 팟에 참여하는 경우, 채팅방에 "~~님이 팟에 참여했습니다"라는 시스템 메시지를 전송해야 했습니다. 저는 참여 로직은 시스템 메시지 전송에는 크게 관심이 없다고 생각했고 비동기로 수행되도록 구현하고 싶었습니다.

public Long participateParty(Long partyId) {
    팟 참여 비즈니스 로직 수행
    ...
	시스템 메시지 전송
}

제가 고민한 방법은 1. @Async를 사용한 메서드를 호출하는 것과 2. 스프링 이벤트를 사용하는 것이었습니다.

1. `@Async`를 사용해 비동기로 메서드 실행하기

Spring에서는 어노테이션만을 사용해 메서드를 비동기로 실행할 수 있도록 지원합니다.

참고로 @Async 어노테이션은 public 메서드에 적용해야 하고 같은 클래스에서 비동기 메서드를 호출하는 것은 비동기로 동작하지 않습니다.

우선 @Configuration이 있는 클래스에 @EnableAsync를 추가해 비동기 처리를 활성화합니다.

@EnableAsync
@Configuration
public class AsyncConfiguration {
}

메시지를 전송하는 메서드에 @Async를 추가해 호출한 스레드 이외의 다른 스레드에서 비동기로 실행되도록 했습니다.

@Component
public class SystemMessageSender {

	@Async
    public void sendParticipateMessage(Party party, User participant) {
    	팟 참여 메시지 전송
    }
}

이제 팟 참여 로직에서 아래와 같이 비즈니스 로직이 모두 실행되고 참여 메시지를 전송하는 메서드를 호출하는 방식으로 작성할 수 있습니다. 비동기로 실행한 덕분에 메시지 전송이 완료되기 전에 participateParty() 메서드는 종료되고 응답을 보낼 수 있습니다. 아무리 메시지 전송 로직이 느려져도 팟 참여 요청은 잘 수행하고 빠르게 응답할 수 있습니다.

public Long participateParty(Long partyId) {
    팟 참여 비즈니스 로직 수행
    ...
    systemMessageSender.sendParticipateMessage(party, authenticatedUser);
}

2. 스프링 이벤트를 사용해 비동기로 이벤트 처리하기

마찬가지로 스프링 이벤트를 사용해 비동기로 실행되도록 구현할 수 있습니다.

이벤트 발행같은 경우에는 ApplicationEventPublisher 구현체 빈을 주입받아 사용하면 됩니다.

@Component
public class ParticipationEventPublisher {

    private final ApplicationEventPublisher eventPublisher;

    public ParticipationEventPublisher(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    public void publishParticipateEvent(Party party, User participant) {
        ParticipateEvent event = new ParticipateEvent(party, participant);
        eventPublisher.publishEvent(event); // 이벤트 발행
    }
}

이벤트를 받아 처리하는 부분도 @EventListener를 사용해 쉽게 구현할 수 있습니다. 여기서 이벤트 처리를 비동기로 수행하는 방법은 이벤트 리스너 메서드에 @Async를 추가하거나 ApplicationEventMulticaster를 커스텀해 빈으로 등록하는 방법이 있습니다.

@Component
public class ParticipationEventSubscriber {

    @EventListener
    public void handleParticipateEvent(ParticipateEvent participateEvent) {
    	팟 참여 메시지 전송
    }
}

ApplicationEventMulticaster 커스텀하기

저는 ApplicationEventMulticaster를 커스텀하는 방법을 선택했습니다. 그 이유는 팟 참여 이벤트 이외의 여러 이벤트 모두 비동기로 처리하고 싶었고, 무엇보다 TaskExecutor를 변경할 수 있기 때문입니다.

ApplicationEventMulticaster를 커스텀하면 모든 이벤트 리스너 메서드에 @Async를 추가하지 않고도 전역적으로 이벤트 처리가 비동기로 실행되도록 할 수 있습니다.

ApplicationEventMulticaster는 아래와 같이 TaskExecutor를 커스텀해 빈으로 등록할 수 있습니다. 그럼 모든 이벤트 리스너는 비동기로 처리되며 해당 TaskExecutor에 의해 실행됩니다.

@Configuration
public class AsynchronousSpringEventConfiguration {

    @Bean
    public ApplicationEventMulticaster applicationEventMulticaster() {
        SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
        TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("SimpleAsync-");
        eventMulticaster.setTaskExecutor(taskExecutor);
        return eventMulticaster;
    }
}

TaskExecutor는 2가지 선택지가 있었습니다.

SimpleAsyncTaskExecutor
ThreadPoolTaskExecutor

나중에 기회가 된다면 TaskExecutor와 관련해 글을 정리해 볼까 합니다!

SimpleAsyncTaskExecutor의 경우, 작업 실행을 요청하면 새로운 스레드를 생성해 작업을 처리합니다. concurrentLimit 변수를 설정해 동시에 실행할 수 있는 작업의 수를 제한할 수도 있습니다.

ThreadPoolTaskExecutor는 이름처럼 스레드 풀에 스레드를 미리 생성해 두고 작업 실행을 요청하면 해당 풀에서 스레드를 꺼내 작업을 처리합니다. 내부에서 java.util.concurrent.ThreadPoolExecutor를 사용하며 corePoolSize와 maxPoolSize, queueCapacity 등의 값을 설정할 수 있습니다.

저는 두 개의 TaskExecutor 중 ThreadPoolTaskExecutor를 선택했습니다.

팟 참여 이벤트 이외에도 팟 나가기 이벤트, 채팅 전송 이벤트 등 다양한 이벤트를 발행하고 처리해야 했습니다. 이런 이벤트들은 매우 빈번히 발생하고 작업 처리 시간이 매우 짧습니다. 그래서 이벤트가 발생할 때마다 새로운 스레드를 생성하는 SimpleAsyncTaskExecutor 보다는 미리 스레드를 생성해 두고 스레드를 재활용할 수 있는ThreadPoolTaskExecutor가 더 적합하다고 생각했습니다.

구성은 아래와 같이 작성했습니다.

@Configuration
public class AsynchronousSpringEventConfiguration {

    @Bean
    public ApplicationEventMulticaster applicationEventMulticaster(
            @Qualifier("eventHandlerTaskExecutor") TaskExecutor taskExecutor) {
        SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
        eventMulticaster.setTaskExecutor(taskExecutor);
        return eventMulticaster;
    }

    @Bean
    public TaskExecutor eventHandlerTaskExecutor() {
        return new ThreadPoolTaskExecutorBuilder()
                .corePoolSize(10)
                .maxPoolSize(50)
                .queueCapacity(200)
                .threadNamePrefix("EventHandler-")
                .build();
    }
}

참고로 ThreadPoolTaskExecutor의 경우에는 스프링 빈으로 등록하지 않고 생성한 객체를 직접 사용하는 경우, initialize() 메서드를 호출해 주지 않으면 ThreadPoolTaskExecutor not initialized 오류가 발생합니다.
ThreadPoolTaskExecutor는 InitializingBean 인터페이스의 afterPropertiesSet() 메서드를 initialize()가 호출되도록 구현하고 있기 때문에(정확히는 ExecutorConfigurationSupport에서) 스프링 빈으로 등록하는 경우에는 자동으로 초기화가 진행됩니다.

이제 이벤트 리스너 메서드에 @Async를 사용하지 않아도 직접 설정한 TaskExecutor가 이벤트를 처리하는 것을 확인할 수 있습니다.

팟 참여 로직에서 비즈니스 로직을 모두 수행하고 이벤트를 발행하는 방식으로 구현하여 메시지 전송 로직이 비동기로 실행되도록 했습니다.

public Long participateParty(Long partyId) {
    팟 참여 비즈니스 로직 수행
    ...
    eventPublisher.publishParticipateEvent(party, authenticatedUser);
}

트랜잭션과 함께 사용하기

하지만 고려해야 할 부분이 한 가지 더 있었습니다.

저는 팟 참여 로직을 하나의 트랜잭션이 수행되도록 하기 위해 @Transactional을 사용하고 있었습니다. 데이터의 정합성과 작업의 원자성을 보장하기 위함이었습니다. 문제는 팟 참여 메시지 전송을 비동기로 실행하면서 발생합니다.

@Transactional // 트랜잭션 적용
public Long participateParty(Long partyId) {
    팟 참여 비즈니스 로직 수행
    ...
	시스템 메시지 전송 // 비동기
}

위의 코드를 실행하면 팟 참여 로직은 트랜잭션 내에서 실행되고 메시지 전송 로직은 트랜잭션 외부에서 다른 스레드에 의해 실행됩니다. 팟 참여 로직에서 메시지 전송 로직에 대한 관심사를 최대한 분리하고 메시지 전송 로직에 관여하지 않도록 했기 때문에 당연하고 의도한 동작입니다.

하지만 트랜잭션이 Commit 되지 않고 예기치 못한 문제가 발생해 트랜잭션이 Rollback 되어 팟 참여 로직이 완료되지 않은 경우에도 메시지 전송 로직은 정상적으로 완료됩니다. 이 경우, 사용자가 팟에 참여되지 않았지만 참여 메시지는 전송되는 문제가 생길 수 있습니다.

트랜잭션 이벤트 리스너

다행히도 Spring에서는 트랜잭션이 성공적으로 완료되었을 때 이벤트를 처리할 수 있도록 지원합니다.

@EventListener 대신 @TransactionalEventListener를 사용하면 이벤트를 발행한 트랜잭션의 커밋이 완료되는 경우에만 이벤트를 처리합니다.

@Component
public class ParticipationEventSubscriber {

    @TransactionalEventListener
    public void handleParticipateEvent(ParticipateEvent participateEvent) {
    	팟 참여 메시지 전송
    }
}

@TransactionalEventListener에는 phase 속성을 통해 이벤트 처리 시점을 지정할 수 있으며 총 4가지 타입이 존재합니다. 기본 설정은 커밋이 완료된 후에 이벤트를 처리하는 동작입니다.

BEFORE_COMMIT - 트랜잭션이 커밋되기 전에 이벤트 처리
AFTER_COMMIT(default) - 트랜잭션이 커밋되어 완료된 후 이벤트 처리
AFTER_ROLLBACK - 트랜잭션이 롤백된 후 이벤트 처리
AFTER_COMPLETION - 트랜잭션이 완료된 후(커밋 또는 롤백) 이벤트 처리

하지만 @TransactionalEventListener를 사용하면 이벤트 리스너 메서드가 비동기로 실행되지 않기 때문에 명시적으로 @Async를 사용해야 합니다. 그냥 @Async를 사용하면 기본 TaskExecutor를 사용하게 되므로 아까 구성한 ThreadPoolTaskExecutor를 통해 실행되도록 value 속성에 "eventHandlerTaskExecutor"를 명시해 주었습니다.

@Component
public class ParticipationEventSubscriber {

	@Async("eventHandlerTaskExecutor")
    @TransactionalEventListener
    public void handleParticipateEvent(ParticipateEvent participateEvent) {
    	팟 참여 메시지 전송
    }
}

팟 참여 이벤트를 발행하고 메서드가 종료되기 전에 예외를 발생하도록 테스트 해보면 이벤트 리스너가 실행되지 않는 것을 확인할 수 있습니다. 또, 이벤트 리스너 메서드에서 스레드를 sleep 시키고 팟 참여 요청을 보냈을 때 이벤트 리스너는 비동기로 실행되기 때문에 빠르게 응답이 오는 것도 볼 수 있습니다.

프로젝트를 진행하며 스프링에서 제공하는 이벤트를 사용해 이벤트를 발행하고 구독하는 방식으로 처리해 봤습니다. 이벤트를 발행하는 주체에서는 이벤트를 구독하는 주체를 전혀 몰라도 된다는 점에서 매력이 있는 것 같습니다. 나중에 동일한 이벤트에 대한 여러 처리가 필요하다면 이벤트 리스너만 추가해 쉽게 확장할 수 있을 것 같습니다. 하지만 이벤트를 발행할 때마다 이벤트 객체를 생성해야 하는 오버헤드가 존재하기도 합니다. 이벤트가 발행될 때마다 메모리에 객체가 생성되고 처리가 완료되면 GC에 의해 정리되는 과정이 빈번하게 발생할 테니까요.

이벤트 리스너 메서드 이름을 handleXxxEvent() 처럼 작성했는데, 어떤 작업이 이루어지는지를 나타낼 수 있도록 네이밍을 해봐야 겠습니다. 아마 스프링이 어노테이션으로 이벤트 리스너를 구성할 수 있도록 한 것에는 이러한 이유도 있지 않았을까요!

References

https://www.baeldung.com/spring-events

https://www.baeldung.com/spring-async

https://docs.spring.io/spring-framework/reference/data-access/transaction/event.html

저작자표시

Spring Events

사용 배경

1. @Async를 사용해 비동기로 메서드 실행하기

2. 스프링 이벤트를 사용해 비동기로 이벤트 처리하기

ApplicationEventMulticaster 커스텀하기

트랜잭션과 함께 사용하기

트랜잭션 이벤트 리스너

References

티스토리툴바

1. `@Async`를 사용해 비동기로 메서드 실행하기