Event Publisher로 조회수 비동기적으로 처리하기 (+ @Async를 빠뜨리면?)

📗 시작하며

프로젝트를 진행하며 조회수 업데이트를 비동기적으로 처리하게 되며 Event Publisher를 공부하게 되었습니다.

해당 글은 Event Publisher를 사용하게 된 이유와 기본적인 사용법을 코드 예시와 함께 설명하고 있습니다.

또한 @TransactionalEventListener와 @EventListener 동작의 차이점을 테스트하던 중 깨달음을 얻은 내용을 담고 있습니다.

바로 코드를 통해 확인해보겠습니다!

@Service
@RequiredArgsConstructor
public class ClubFacade {
	private final ClubService clubService;
	private final ClubStatisticsService clubStatisticsService;

	@Transactional
	public ClubInfo readClub(String clubToken) {
		var clubInfo = clubService.readClub(clubToken);
        clubStatisticsService.increaseClubVisitCount(clubToken);
		return clubInfo;
	}
	/* 
  * 생략
  */
}

동호회를 조회하는 서비스 코드 일부입니다. 동호회를 조회한 다음, 동호회 통계 서비스의 increaseClubVisitCount 메서드를 실행해 조회수를 1 올리고 있습니다.

📍 +) 조회수의 중요성

참고로, 저희 서비스의 경우 조회수를 사용자에게 직접적으로 보여주지 않고 있습니다. 인기 있는 동호회를 정렬할 때 내부적으로 사용하는 지표인데요.

이러한 이유로, 조회수의 정합성이 아주 중요하다고 판단하지 않았습니다. 즉, 10만 명의 사용자가 동호회를 조회했지만, 조회수가 10만보다 조금 적더라도, 이는 서비스에 큰 영향을 미치지 않습니다.

따라서 동호회 조회 시 조회수가 일부 올라가는데 실패하더라도, 사용자가 동호회를 정상적으로 조회할 수 있는 것이 더 중요한 요구사항으로 작용했습니다.

개발 중인 서비스에서 조회수가 어떤 역할을 하고 있는지, 또한 실시간성과 데이터 정합성이 어느 정도로 보장되어야 하는지를 고민해보면 좋을 것 같습니다!

🤔 문제점

1. 강결합

지금은 동호회와 관련된 서비스가 동호회 통계 서비스와 강하게 결합이 되어있습니다. 동호회를 조회할 때 ClubStatisticsService에 꼭 의존해야 할까요?

2. 조회수 쓰기 작업 실패 시 동호회 조회도 실패

조회수를 올리는 기능이 동기적으로 처리되고 있으며, 동호회 조회와 조회수 쓰기 작업이 하나의 트랜잭션으로 묶여 있습니다. 아래의 상황을 생각해보겠습니다.

조회수를 올리는 로직에서 예외가 발생한다면?

동호회 조회에는 성공했지만, 조회수를 올리는데 실패했다고 가정해보겠습니다. 이를 실패로 간주해야 할까요?
현재 동호회 조회와 조회수 쓰기 작업이 @Transactional로 묶여 있기 때문에 하나가 실패하면 나머지 작업도 실패한 것으로 간주됩니다.
조회수를 올리는데 실패했다고 동호회 조회 API가 에러를 응답해야 할까요?

앞서 이야기했듯이, 저희 서비스의 경우 조회수가 100% 정확하지 않더라도 사용자 경험에 큰 영향을 미치지 않습니다. 따라서 조회수가 실제와는 조금 다를 수도 있다는 점은 감수할 수 있다고 판단했습니다.

📮 Application Event Publisher

스프링에서 제공하는 Event 기능을 사용해 조회수를 올리는 작업을 비동기적으로 처리해 위의 두 가지 문제점을 해결할 수 있습니다.

1. Event

이벤트와 관련된 데이터를 저장하는 Placeholder, 또는 DTO 객체를 아래와 같이 작성합니다. 동호회 조회 이벤트의 경우, 동호회 UUID를 필요로 합니다.

** 참고로, Event는 아래와 같이 일반 객체로 만들어도 무방합니다. Spring Framework 4.2 이전에는 이벤트 객체가 반드시 ApplicationEvent를 상속해야 했습니다. 4.2 이후부터 publishEvent(Object event) 메서드가 오버로드되어 일반 객체를 이벤트 객체로 사용할 수 있습니다.

@Getter
@RequiredArgsConstructor
public class ReadClubEvent {

	private final String clubToken;
}

2. Listener

리스너는 @EventListener 어노테이션을 붙이거나 ApplicationListener 인터페이스를 정의해 구현할 수 있습니다. 아래 예시에서는 간단하게 어노테이션을 붙여 Event Listener를 구현했습니다.

파라미터에 정의한 타입의 이벤트가 발생하면 @EventListener 어노테이션을 붙인 퍼블릭 메서드가 동작합니다.

@Component
@RequiredArgsConstructor
public class ReadClubEventHandler {

	private final ClubStatisticsService clubStatisticsService;

	@EventListener
	public void readClubEventListener(ReadClubEvent event) {
		clubStatisticsService.increaseVisitedClubCount(event.getClubToken());
	}
}

이때 한 가지 기억해야 할 점은, 스프링은 default일 때 이벤트를 동기적(Synchronous)으로 처리합니다.

3. Publisher

이벤트와 이벤트 리스너가 준비되었으니, 이제 이벤트를 발행할 차례입니다.

3.1. ApplicationEventPublisher

ApplicationEventPublisher라는 함수형 인터페이스를 사용해 이벤트를 발행할 수 있는데요, 아래와 같이 publishEvent라는 추상 메서드를 포함하고 있습니다.

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

    void publishEvent(Object event);
}

publishEvent 메서드를 통해 이벤트가 발행되면, Spring은 Application Context에 등록된 리스너 중 해당 이벤트와 매칭되는 리스너를 찾아 호출합니다.

해당 함수형 인터페이스는 ApplicationContext 를 구현한 AbstractApplicationContext 클래스에 구현되어 있습니다.

public interface ApplicationContext extends ApplicationEventPublisher /*, ... 생략*/ {

따라서 별도로 구현하지 않고, 간단하게 의존성을 주입받아 publishEvent메서드를 사용할 수 있습니다.

3.2. 코드에 적용

ApplicationEventPublisher에 대한 의존성을 추가하고, publishEvent 메서드를 호출해 이벤트를 발행합니다.

@Service
@RequiredArgsConstructor
public class ClubFacade {
	private final ClubService clubService;
    private final ApplicationEventPublisher eventPublisher;

	@Transactional
	public ClubInfo readClub(String clubToken) {
		var clubInfo = clubService.readClub(clubToken);
        eventPublisher.publishEvent(new ReadClubEvent(clubToken));
		return clubInfo;
	}
	/* 
  * 생략
  */
}

👛 Before & After

Before

@Service
@RequiredArgsConstructor
public class ClubFacade {
	private final ClubService clubService;
	private final ClubStatisticsService clubStatisticsService;

	@Transactional
	public ClubInfo readClub(String clubToken) {
		var clubInfo = clubService.readClub(clubToken);
        clubStatisticsService.increaseClubVisitCount(clubToken);
		return clubInfo;
	}
	/* 
  * 생략
  */
}

After

@Service
@RequiredArgsConstructor
public class ClubFacade {
	private final ClubService clubService;
    private final ApplicationEventPublisher eventPublisher;

	@Transactional
	public ClubInfo readClub(String clubToken) {
		var clubInfo = clubService.readClub(clubToken);
        eventPublisher.publishEvent(new ReadClubEvent(clubToken));
		return clubInfo;
	}
	/* 
  * 생략
  */
}

먼저, 더이상 ClubStatisticsService에 의존하고 있지 않습니다. 동호회 서비스와 동호회 통계 서비스 간의 결합도가 낮아졌습니다.

또한, 동호회 읽기 작업은 동호회 조회수를 올리는 쓰기 작업과 무관하게 동작합니다. 동호회 조회수를 올릴 때 예외가 발생하더라도, 동호회 읽기 API는 성공적으로 요청을 처리합니다.

☔️ 테스트.. 어라라?!

두 가지 테스트를 진행해보려고 합니다.

예상한 시나리오 1
이벤트 핸들러에서 예외가 발생하더라도, 동호회 조회 API는 정상 응답할 것이다.

동호회 조회수를 올릴 때 문제가 발생하더라도, 동호회 조회는 성공해야 한다는 요구사항이 있습니다. 따라서 Event Listener 내부에 예외를 발생하는 코드를 작성했습니다. 이벤트를 처리하다가 예외가 발생해도, 동호회 조회를 하는 API는 성공해야 합니다.

@Component
@RequiredArgsConstructor
public class ReadClubEventHandler {

	private final ClubStatisticsService clubStatisticsService;

	@EventListener
	public void readClubEventListener(ReadClubEvent event) {
		clubStatisticsService.increaseVisitedClubCount(event.getClubToken());
        // 예외를 발생시킨다.
		throw new BadmintonException(ErrorCode.CLUB_NOT_EXIST);
	}
}

그러나, 예상했던 것과는 다르게 에러를 응답하고 있습니다.

예상한 시나리오 2
@EventListener를 사용했기 때문에 트랜잭션 실패에도 불구하고 이벤트가 실행되어 조회수가 늘어나 있을 것이다.

@TransactionalEventListener를 붙인 이벤트 리스너는 기본적으로 트랜잭션 커밋 후 실행됩니다.

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)  // Default 옵션

트랜잭션 커밋 이후 리스너가 실행되며, 만약 트랜잭션이 실패하면 리스너는 실행되지 않습니다. 즉, @Transactional 내 모든 작업이 성공해야 이벤트 리스너가 실행됩니다.

이 둘의 차이점을 느껴보고자 먼저 @EventListener를 사용해 테스트를 진행했습니다. 아래와 같이 이벤트를 발행한 다음 예외를 발생하는 코드를 추가했습니다.

@Transactional
	public ClubCreateInfo createClub(ClubCreateCommand createCommand, String memberToken) {
		ClubCreateInfo clubCreateInfo = clubService.createClub(createCommand);
		clubMemberService.clubMemberOwner(memberToken, clubCreateInfo);
		eventPublisher.publishEvent(new CreateClubEvent(clubCreateInfo));
        // 예외를 발생시킨다. 
		throw new BadmintonException(ErrorCode.CLUB_NOT_EXIST);
	}

예상했던 것과 같이 트랜잭션이 종료되기 전이지만, 이벤트 리스너 내부의 코드로 진입하는 것을 확인했습니다.

다만, 최종적으로 롤백되어 조회수가 늘어나지 않았습니다.

(기존 조회수 110, API 호출 이후 조회수 여전히 110)

‼️ @Async를 안 붙이면 Spring은 이벤트를 동기적으로 처리

EventHandler에 @Async 어노테이션을 붙이지 않아 조회수를 증가하는 로직이 동기적으로 처리되고 있었습니다.

스프링은 default 옵션에서 이벤트를 동기적으로 처리하며, 이때 이벤트 리스너는 동일한 스레드에서 이벤트를 처리합니다.
즉, 동호회를 조회하고, 조회수를 올리는 두 가지 로직이 동일 스레드에서 실행됩니다.

따라서 동일 스레드 내에 예외가 발생했을 때 (1) 이를 실패로 간주해 에러를 응답하고, (2) 조회수 쓰기 작업은 롤백된 것입니다.

아래는SimpleApplicationEventMulticaster 클래스의 코드 일부입니다. 이벤트가 발생하면 이를 적합한 리스너에게 전달하며, 비동기 처리를 위해 TaskExecutor를 사용합니다.

public void multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType) {
        ResolvableType type = eventType != null ? eventType : ResolvableType.forInstance(event);
        Executor executor = this.getTaskExecutor();
        Iterator var5 = this.getApplicationListeners(event, type).iterator();

        while(true) {
            while(var5.hasNext()) {
                ApplicationListener<?> listener = (ApplicationListener)var5.next();
                /* Executor가 설정되어 있고 리스너가 비동기를 지원하면 */
                if (executor != null && listener.supportsAsyncExecution()) {
                    try {
                        executor.execute(() -> {
                        /* 이벤트를 리스너에게 전달 */
                            this.invokeListener(listener, event);
                        });
                    } catch (RejectedExecutionException var8) {
                        this.invokeListener(listener, event);
                    }
                } else {
                /* 설정된 Executor가 없다면 동기적으로 실행된다 */
                    this.invokeListener(listener, event);
                }
            }

            return;
        }
    }
        /* 생략 */

Executor가 없으면, 이벤트는 동기적으로 실행됩니다. 즉, 이벤트 리스너가 현재 스레드에서 호출됩니다.

반면, Executor가 설정되어 있다면, 이벤트는 비동기적으로 실행됩니다. 즉, 이벤트 리스너가 별도의 스레드에서 호출됩니다.

따라서, @Async 어노테이션을 붙여야 이벤트가 비동기적으로 처리됩니다. 이때 리스너는 이벤트를 별개의 스레드에서 처리합니다.

지금은 이벤트가 동기적으로 실행되고 있고 @Transactional로 묶여 있기 때문에 트랜잭션에 참여한 스레드에서 예외가 발생할 때 전부 롤백합니다. 그래서 조회수 증가 로직이 실행되었음에도 불구하고, 롤백이 된 것이죠.

🧵 동기적으로 실행 시 이벤트 리스너가 동일 스레드에서 동작

앞서 말했듯이, 이벤트 리스너가 동일 스레드에서 돌고 있으며, 동일 스레드 내에서 예외가 발생했으니 롤백시키고 있습니다.

동기적으로 이벤트를 실행할 때 정말 동일 스레드인지 직접 확인해보도록 하겠습니다.

먼저 아래와 같이 @Async 어노테이션 없이 기존처럼 이벤트를 동기적으로 처리해보도록 하겠습니다.

@Component
@RequiredArgsConstructor
public class ReadClubEventHandler {

	private final ClubStatisticsService clubStatisticsService;
	
	@EventListener
	public void readClubEventListener(ReadClubEvent event) {
		System.out.println("**** 이벤트 리스너 스레드 이름: " + Thread.currentThread().getName());
		clubStatisticsService.increaseVisitedClubCount(event.getClubToken());
	}
}

이벤트 리스너가 동호회를 조회하는 스레드와 동일한 스레드임을 확인했습니다.

이제 이벤트 리스너에 @Async 어노테이션을 붙여 이벤트를 비동기적으로 처리해보도록 하겠습니다.

@Component
@RequiredArgsConstructor
public class ReadClubEventHandler {

	private final ClubStatisticsService clubStatisticsService;

	@Async
	@EventListener
	public void readClubEventListener(ReadClubEvent event) {
		System.out.println("**** 이벤트 리스너 스레드 이름: " + Thread.currentThread().getName());
		clubStatisticsService.increaseVisitedClubCount(event.getClubToken());
	}
}

@SpringBootApplication
@EnableAsync
public class BadmintonApplication {
	public static void main(String[] args) {
		SpringApplication.run(BadmintonApplication.class, args);
	}
}

이벤트 리스너가 별개의 스레드에서 호출되는 것을 확인할 수 있었습니다.

이제 이벤트가 비동기적으로 처리되고 있으니, 앞서 예상했던 두 가지 시나리오를 다시 확인해보도록 하겠습니다.

예상한 시나리오 1
이벤트 핸들러에서 예외가 발생하더라도, 동호회 조회 API는 정상 응답할 것이다.

동호회 조회수를 올리는 로직에서 예외가 발생하지만, 이는 비동기적으로 처리되며 별도의 스레드에서 동작합니다. 따라서 동호회를 조회하는 API는 이벤트 리스너의 동작과는 무관하게 성공합니다.

예상한 시나리오 2
@EventListener를 사용했기 때문에 트랜잭션 실패에도 불구하고 이벤트가 실행되어 조회수가 늘어나 있을 것이다.

@TransactionalEventListener가 아닌 @EventListener를 사용했기 때문에, 트랜잭션이 실패했음에도 불구하고 이벤트가 처리되는 것을 확인했습니다. (기존에 조회수가 110이었고, 현재 1 증가한 상황)

🛤️ @TransactionalEventListener

위에서 @EventListener를 사용하면, 트랜잭션 실패 시에도 이와 무관하게 이벤트가 처리되는 것을 확인할 수 있었습니다.

다만, 조회수 읽기에 실패하면 조회수 또한 올라가지 않아야 한다는 요구사항이 있습니다. 이를 보장하려면, @TransactionalEventListener를 사용할 수 있습니다!

@EventListener 대신 @TransactionalEventListener를 사용하면 트랜잭션이 커밋되고 난 이후에 이벤트가 처리됩니다.

트랜잭션을 커밋하기 전에 예외를 발생시켜 이벤트가 정말 실행되지 않는지 확인해보도록 하겠습니다.

@Component
@RequiredArgsConstructor
public class ReadClubEventHandler {

	private final ClubStatisticsService clubStatisticsService;

	@Async
	@TransactionalEventListener
	public void readClubEventListener(ReadClubEvent event) {
		System.out.println("**** 이벤트 리스너 스레드 이름: " + Thread.currentThread().getName());
		clubStatisticsService.increaseVisitedClubCount(event.getClubToken());
	}
}

동호회 조회와 @Transactional로 함께 묶여 있는 블록 내에 예외를 던지는 코드를 작성했습니다.

@Transactional
	public ClubDetailsInfo readClub(String clubToken) {
		clubService.readClub(clubToken);
		eventPublisher.publishEvent(new ReadClubEvent(clubToken));
		throw new BadmintonException(ErrorCode.CLUB_NOT_EXIST);
	}

위에서 @EventListener를 사용했을 때는, 트랜잭션 커밋 여부와 상관없이 이벤트가 처리되었습니다.

반면, @TransactionalEventListener의 경우, 이벤트가 발행은 되지만, 트랜잭션 실패 시 이벤트가 처리되지 않는 것을 확인할 수 있었습니다.

이벤트가 발행은 되었으나, 트랜잭션이 실패했기 때문에 이벤트를 처리하지 않았습니다. 따라서 아래와 같이 조회수는 여전히 110입니다.

이제 동호회 조회에 실패하면, 동호회 조회수는 올라가지 않습니다! 😁

👀 헷갈렸던 부분

처음에는 @TransactionalEventListener 사용 시 트랜잭션이 실패했을 때 이벤트 처리가 롤백되는 것이라고 이해했습니다. 이는 틀린 말이며, 트랜잭션이 실패하면 이벤트 리스너가 아예 이벤트를 처리하지 않습니다. (혹시 저처럼 헷갈리신 분이 계실까봐 . . . 😅)

🔮 개선할 점

지금은 동호회 GET 요청이 들어왔을 때, 동일 사용자인지 등을 확인하지 않고 조회수를 올리고 있습니다. 동일 사용자에 대해서는 조회수를 1만 올리도록 제한하거나, 세션 정보를 활용해 개선할 예정입니다.

또한 레디스를 활용해 조회가 있을 때마다 DB에 업데이트를 하지 않고, 특정 시간이 지나거나 어느 정도 데이터가 쌓였을 때 DB에 업데이트를 하도록 수정하려고 합니다. 서비스에서 조회수의 실시간성이 아주 중요하다고 판단하지 않았기 때문입니다.

🎄 마치며

조회수를 비동기적으로 올리기 위해 Event Publisher를 공부하며 스레드에 대해서도 추가적인 공부를 할 수 있었습니다. 이제 조회수를 올리는 작업은 동호회 조회와는 별개의 스레드에서 비동기적으로 처리됩니다!

다만, 스프링에서 지원하는 Event는 동일 프로세스 내에서만 동작합니다. 스프링 이벤트 시스템이 ApplicationContext를 기반으로 동작하기 때문인데, 이러한 이유로 분산 환경에서는 별도의 메시징 시스템이 필요합니다. 카프카, RabbitMQ 등이 있으며 아직 공부해보지 못해서 추후에 적용해볼 계획입니다!

모두 해피 연말 보내세요 😁

📚 참고 자료

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