[Spring] 스프링 이벤트, @TransactionalEventListener

 

1. 스프링 이벤트(Spring Event)를 사용하는 이유

스프링 이벤트(Spring Event)를 사용하는 이유는 서비스 간의 의존성을 줄이기 위해서이다.

스프링 이벤트를 사용하기 전에는 OrderService가 (1)사용자의 주문 요청을 처리하고 (2)푸시 메시지 발송 및 (3)메일 전송을 처리한다. OrderService는 PushService와 MailService를 의존하게 된다.

반면에 스프링 이벤트를 사용하면 OrderService는 사용자의 주문 요청을 처리하는 책임만 갖고, 스프링 이벤트가 푸시 메시지 발송과 메일 전송을 처리한다. OrderService는 주문 요청만 처리하기 때문에 푸시 메시지 발송 및 메일 전송에 대한 책임을 갖지 않는다.

 

2. 스프링 이벤트(Spring Event) 구성 요소

스프링 이벤트는 이벤트 객체(Event)와 이벤트 발행자(EventPublisher) 그리고 이벤트 리스너(EventListener)이 존재한다. EventPublisher에 의해 실행된 Event를 EventListener가 처리한다.

2-1. 이벤트 객체 (Event)

발생할 이벤트에 대한 정보를 담는다.

@AllArgsConstructor
@Getter
public class SectionCacheUpdateEvent {

    private Long retrospectiveId;
    private Section section;
    private User user;

}

2-2. 이벤트 리스너 (EventListener)

@TransactionalEventListener을 사용하여 Event를 핸들링한다. 핸들러 대상은 메서드의 매개변수로 전달되는 Event 객체이다.

아래의 경우 SectionCacheUpdateEvent 이벤트 객체를 이벤트 리스너가 처리한다.

@Service
@RequiredArgsConstructor
@Slf4j
public class SectionCacheListener {

    private final CacheRepository<List<GetSectionsResponseDto>> cacheRepository;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void updateCacheWithNewSection(SectionCacheUpdateEvent event) {
        String cacheKey = String.format("%s::%d", cacheRepository.getCacheKey(),
            event.getRetrospectiveId());

        try {
            // 캐싱된 데이터를 가져온다.
            List<GetSectionsResponseDto> cachingData = cacheRepository.getCacheDate(cacheKey);

            // 캐싱된 값이 존재할 경우 생성된 회고 카드를 추가한다.
            if (cachingData != null && !cachingData.isEmpty()) {
                cachingData.add(GetSectionsResponseDto.from(event.getSection(), event.getUser()));
                cacheRepository.saveCacheData(cacheKey, cachingData);
            }
        } catch (Exception ex) { // TODO 커스텀 예외 생성한다.
            // 캐싱되어 있는 데이터를 삭제한다.
            cacheRepository.deleteCacheData(cacheKey);
            log.error("캐싱 데이터 갱신 중 오류가 발생했습니다.", ex);
        }
    }
}

2-3. 이벤트 발행자 (EventPublisher)

eventPublisher.publishEvent(Event)를 호출하여 이벤트를 발생시킬 수 있다.

아래는 SectionCacheUpdateEvent 이벤트 객체를 생성하여 이벤트를 발생시키고 있다.

@Service
@RequiredArgsConstructor
@Slf4j
public class SectionService {

    private final SectionRepository sectionRepository;
    private final RetrospectiveRepository retrospectiveRepository;
    private final TemplateSectionRepository templateSectionRepository;
    private final LikesRepository likesRepository;
    private final TeamRepository teamRepository;
    private final ActionItemRepository actionItemRepository;
    private final UserRepository userRepository;
    private final KudosTargetRepository kudosRepository;
    private final NotificationRepository notificationRepository;
    private final ApplicationEventPublisher eventPublisher;

    // 회고 카드 생성 API
    @Transactional
    public CreateSectionResponseDto createSection(User user, CreateSectionDto request) {
        Retrospective findRetrospective = getRetrospective(request.getRetrospectiveId());
        TemplateSection findTemplateSection = getTemplateSection(request.getTemplateSectionId());

        /**
         * 회고 템플릿이 일치하는지 확인한다.
         * 회고 템플릿이 일치하지 않으면 예외를 발생시킨다.
         */
        validateTemplateMatch(findRetrospective, findTemplateSection);

        // 회고 카드 생성
        Section createSection = createSection(request.getSectionContent(), findTemplateSection,
            findRetrospective, user);
        Section saveSection = sectionRepository.save(createSection);

        // 캐싱된 데이터에 새로운 회고 카드를 추가한다.
        eventPublisher.publishEvent(
            new SectionCacheUpdateEvent(request.getRetrospectiveId(), saveSection, user));

        // Entity를 Dto로 변환하여 반환한다.
        return convertCreateSectionResponseDto(request, createSection);
    }
    
}

 

3. @TransactionalEventListener

스프링에서는 @EventListener과 @TransactionalEventListener가 존재한다. @EventListener의 경우 event를 publishing 하는 경우에 바로 이벤트를 발생시킨다. 반면에 @TrasnactionalEventListner는 publishing 시점에 바로 이벤트를 발생시키는 것이 아닌, 트랜잭션의 상태에 따라 이벤트를 핸들링할 수 있다. 따라서 트랜잭션과 연계된 이벤트를 처리하기 위해서는 @TransactionalEventListner를 사용해야 한다.

3-1. @TransactionalEventListner 속성

@TransactionalEventListener(phase = 속성), phase 속성을 통해 이벤트 리스너가 실행될 시점을 지정할 수 있다.

• AFTER_COMMIT(기본) : 트랜잭션이 commit 된 경우에만 이벤트를 실행한다.
• AFTER_ROLLBACK : 트랜잭션이 rollback된 경우에만 이벤트를 실행한다.
• AFTER_COMPLETION : 트랜잭션이 마무리 되었을 때(commit or rollback) 이벤트를 실행한다.
• BEFORE_COMMIT : 트랜잭션 commit 이전에 이벤트를 실행한다.

3-2. @TransactionalEventListner 특징

스프링 이벤트는 기본적으로 동기 방식으로 동작한다. 즉, 이벤트가 발생하면 해당 이벤트를 처리하는 리스너가 즉시 실행되며, 이벤트를 발생시킨 메서드는 리스너의 작업이 완료될 때까지 기다리게 된다. 그러나 @TransactionalEventListener의 경우에는 트랜잭션의 상태에 따라 이벤트 리스너가 실행되는 시점을 제어할 수 있다. 예를 들어, TransactionPhase.AFTER_COMMIT을 사용하면 트랜잭션이 성공적으로 커밋된 후에만 이벤트 리스너가 호출됩니다. 이로 인해 트랜잭션이 정상적으로 완료되기 전에는 이벤트 리스너가 실행되지 않으며, 트랜잭션이 실패하면 이벤트 리스너도 실행되지 않는다.

 

4. EventListener에서 예외가 발생할 경우

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)의 경우 이벤트를 발생시킨 트랜잭션이 commit되어야 이벤트 리스너가 이벤트를 처리한다. 그런데 DB에 commit은 정상적으로 수행되었으나 이벤트 리스너에서 예외가 발생할 수도 있을 것이다. 이런 경우에는 예외를 잡아서 적절히 처리를 해야 한다.

아래의 경우 이벤트 리스너가 이벤트를 처리하다가 예외를 발생하면 캐시를 무효화하도록 하였다. 예외가 발생한 경우 캐시 데이터를 무효화하거나, 실패한 작업을 재시도할 수 있는 로직을 추가하는 것이 좋다.

@Service
@RequiredArgsConstructor
@Slf4j
public class SectionCacheListener {

    private final CacheRepository<List<GetSectionsResponseDto>> cacheRepository;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void updateCacheWithNewSection(SectionCacheUpdateEvent event) {
        String cacheKey = String.format("%s::%d", cacheRepository.getCacheKey(),
            event.getRetrospectiveId());

        try {
            // 캐싱된 데이터를 가져온다.
            List<GetSectionsResponseDto> cachingData = cacheRepository.getCacheDate(cacheKey);

            // 캐싱된 값이 존재할 경우 생성된 회고 카드를 추가한다.
            if (cachingData != null && !cachingData.isEmpty()) {
                cachingData.add(GetSectionsResponseDto.from(event.getSection(), event.getUser()));
                cacheRepository.saveCacheData(cacheKey, cachingData);
            }
        } catch (Exception ex) { // TODO 커스텀 예외 생성한다.
            // 캐싱되어 있는 데이터를 삭제한다.
            cacheRepository.deleteCacheData(cacheKey);
            log.error("캐싱 데이터 갱신 중 오류가 발생했습니다.", ex);
        }
    }
}