스프링 이벤트를 활용해 로직간 강결합을 해결하는 방법

EP·2023년 3월 24일
EventListenerSpringTransactionalEventListenerasyncevent drivent architecturekotlintransactional
10

Kotlin + Spring

목록 보기
9/9

Overview

애플리케이션 로직을 설계하면 한 번의 요청에 의해 2가지 이상의 기능을 동작해야하는 경우가 생깁니다. 이 다수의 기능을 하나의 메서드에서 코드로 구현하면 기능과 기능이 강결합(Tight Coupling)이 됩니다. 각 로직을 분리해서 관리하기도 어렵고 특정 기능의 문제가 발생했을 때 문제를 처리하는 로직이 섞이게 됩니다. 간단한 요구사항으로 예시를 들겠습니다.

서비스의 사용자가 회원가입 요청을 하면 사용자의 데이터를 DB에 저장하고 가입 축하 메일을 발송합니다. 가입 축하 메일의 발송 내역은 기록합니다.

사용자가 회원가입 데이터를 입력하여 요청을 보내고 해당 데이터가 정상적이면 DB에 저장을 완료하고 입력한 이메일로 메일을 발송 로직입니다. 이를 간단한 코드로 나타내면 다음과 같습니다.

@Service
@Transactional
class MemberService(
    private val memberRepository: MemberRepository,
    private val mailService: MailService
) {

    fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

        // 1. 회원 등록
        val savedMember = register(registerMemberRequestData)

        // 2. 가입 축하 메일 발송 (메일 발송 기록은 db에 저장)
        mailService.sendSuccessRegisteredMemberMail(savedMember.id, savedMember.email)

        return RegisterMemberResponseData(memberId = savedMember.id)
    }

    private fun register(requestData: RegisterMemberRequestData): Member {
        val newMember = Member(nickname = requestData.nickname, email = requestData.mail)
        return memberRepository.save(newMember)
    }
}

@Service
@Transactional
class MailService(
    private val emailSender: EmailSender,
    private val emailSendHistoryRepository: EmailSendHistoryRepository
) {

    fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String) {
        val successRegisteredMEmber = SuccessRegisteredMemberMessageGenerator.generate(memberId)
        emailSender.send(successRegisteredMEmber, emailAddress)

        val emailSendHistory = EmailSendHistory(
            sendAt = LocalDateTime.now(),
            message = successRegisteredMEmber,
            targetId = memberId,
            type = EmailType.MEMBER_REGISTER_SUCESS
        )
        emailSendHistoryRepository.save(emailSendHistory)
    }
}

회원을 등록하고 이후에 가입 축하 메일을 발송하는 간단한 코드입니다. 이 코드에는 어떤 문제(혹은 개선해야할 부분)가 있을까요?

첫 번째 문제는 기능간의 의존성입니다. 회원 등록 기능과 메일 전송 기능은 각각 MemberService, MailService 로 분리가 되어있어 보이지만 사실 기능간의 강결합이 발생됩니다. 만약 MailService이 별도의 모듈 혹은 시스템으로 분리해야할 때 MemberService의 코드를 변경해야만 했습니다.

두 번째 문제는 트랜잭션의 경계입니다. 해당 예제는 Spring Framework의 @Transactional 을 활용하여 트랜잭션을 관리합니다. registerProcess() 메서드와 sendSuccessRegisteredMemberMail() 메서드는 각각의 트랜잭션 전파 옵션이 default(REQUIRED)이므로 병합이 되어 단일한 트랜잭션이됩니다. 회원가입을 정상적으로 성공하고 메일을 발송했을 때, 메일 발송 기능에서 Exception이 발생하면 상위 메서드인 registerProcess 트랜잭션이 롤백됩니다. 따라서 주요 기능인 회원가입이 성공적으로 완료해도 메일 발송 로직에 오류가 생기면 회원가입이 실패하게 됩니다. 이메일 기능은 별도의 외부 모듈과 내역을 저장하는 로직이 함께 있기 때문에 에러가 발생할 가능성이 높습니다. 트랜잭션을 분리할 필요가 있습니다.

세 번째는 동기적인 방식의 성능 문제입니다. 회원가입과 이메일 발송은 동기적으로 처리할 필요가 없는 로직입니다. 또한 이메일 발송 서비스는 외부 기능을 사용하는 경우가 많아 성능을 보장할 수 없습니다. 가령 회원가입 성공하는데 200ms가 소요되었는데 이메일 발송 로직이 3000ms가 걸리는 기능이라고 가정하겠습니다. 그러면 사용자가 응답을 받는데 대략 3200ms가 넘는 시간이 소요됩니다. 하지만 회원가입 요청은 이메일 발송을 기다릴 필요도 성공 여부를 알 필요도 없습니다. 따라서 이 기능을 비동기로 처리하는 로직이 필요합니다.

네 번째 문제는 이메일의 발송 여부입니다. 회원가입 메서드를 호출했을 때 트랜잭션이 시작되며 메서드의 구현부의 마지막에 도달할 때까지 예외가 없을 때 commit 명령어를 날립니다. 하지만 commit을 할 때 db 제약조건에 데이터가 맞지 않아 DataIntegrityViolationException 와 같은 에러가 발생하고 commit이 실패합니다. 이 경우 에러가 발생했으므로 실패로 응답을 주면 되지만 이 로직에선 이미 메일은 발송이 된 이후입니다. 이메일 발송은 commit이 성공한 것을 확인한 뒤 발송을 해야 합니다.

이 글에서는 이러한 문제점을 해결하기 위해 다양한 방법을 시도해볼생각입니다.

테스트 코드

@Test
fun `case 1 - 기본 예제`() {
    val memberRequestData = RegisterMemberRequestData(nickname = "ep", email = "ep@email.com")
    val response = memberService.registerProcess(memberRequestData)
    assertThat(response).isNotNull

    val newMember = memberRepository.findById(response.memberId)
    val newEventSendHistory = emailSendHistoryRepository.findById(1L)

    assertThat(newMember).isNotNull
    assertThat(newEventSendHistory).isNotNull
}

해결해야할 문제

  1. 기능간의 의존성
  2. 트랜잭션의 결합
  3. 성능적으로 비효율적인 동기적인 로직
  4. DB 결과와 무관하게 발송되는 이메일

Transaction 분리

예외가 발생했을 때 결합된 트랜잭션에서 발생하는 문제에 대해 먼저 알아보겠습니다. 앞에서 봤던 2번째 문제의 이유로 트랜잭션이 결합되어있습니다. 만약 이메일 발송 로직에서 예외가 발생하면 어떤일 이 벌어질까요?

fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String) {
    ...
    throw RuntimeException("send email exception")
}

임의로 예외 로직을 추가해줬습니다.

fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

				...

        // 2. 가입 축하 메일 발송 (메일 발송 기록은 db에 저장)
        try {
            mailService.sendSuccessRegisteredMemberMail(savedMember.id, savedMember.email)
        } catch (e: RuntimeException) {
            println("catch runtime exception")
        }

        return RegisterMemberResponseData(memberId = savedMember.id!!)
    }

밖에서 try-catch문으로 예외를 잡아주었습니다. 테스트를 실행시키면

org.springframework.transaction.UnexpectedRollbackException: Transaction silently rolled back because it has been marked as rollback-only

위와같은 에러가 발생합니다. try-catch로 예외를 잡아줬지만 sendSuccessRegisteredMemberMail() 메서드를 실행시키면서 생성된 별도의 트랜잭션에서 예외가 발생해서 rollback mark가 묻었고 이 트랜잭션이 병합되면서 모든 트랜잭션이 롤백이 된 상황입니다. 이 경우 트랜잭션을 분리해야 합니다. 이 예외에 대한 자세한 이야기는 제 다른 게시글인 ‘@Transactional 상황별 commit, rollback 전략’(case-8)에서 확인하실 수 있습니다.

트랜잭션 분리를 하기 위해 2가지 방법을 고려해보았습니다. 첫 번째는 Facade 객체를 만들어 각각 개별 트랜잭션을 만들어 처리하는 방법과 두 번째로 트랜잭션의 전파 옵션을 REQUIRES_NEW로 설정해서 트랜잭션을 분리하는 방법입니다.

퍼사드(Facade) 패턴은 하위 객체의 기능의 조합하여 사용하는 패턴이라고 생각할 수 있습니다. 다양한 하위 서비스의 집합체이죠. 제가 정의한 퍼사드 객체에는 @Transactional설정을 넣지 않았습니다. 각자의 기능이 개별 트랜잭션으로 기능하길 원하는 의도였습니다.

@Component
class RegisterMemberFacade(
    private val memberService: MemberService,
    private val mailService: MailService
) {
    fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

        // 1. 회원 등록
        val savedMember = memberService.register(registerMemberRequestData)

        // 2. 가입 축하 메일 발송 (메일 발송 기록은 db에 저장)
        try {
            mailService.sendSuccessRegisteredMemberMail(savedMember.id, savedMember.email)
        } catch (e: RuntimeException) {
            println("catch runtime exception")
        }

        return RegisterMemberResponseData(memberId = savedMember.id!!)
    }
}
@Service
@Transactional
class MemberService(
    private val memberRepository: MemberRepository
) {
    fun register(requestData: RegisterMemberRequestData): Member {
        val newMember = Member(nickname = requestData.nickname, email = requestData.email)
        return memberRepository.save(newMember)
    }
}
@Service
@Transactional
class MailService(
    private val emailSender: EmailSender,
    private val emailSendHistoryRepository: EmailSendHistoryRepository
) {

  fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String) {
      ...
      throw RuntimeException("send email exception")
  }
}

메일 서비스는 여전히 오류를 발생시키고 있습니다. 테스트 결과는 어떻게 나올까요?

@Test
fun `case 1 - 기본 예제`() {
    val memberRequestData = RegisterMemberRequestData(nickname = "ep", email = "ep@email.com")
    val response = registerMemberFacade.registerProcess(memberRequestData)
    assertThat(response).isNotNull

    val newMember = memberRepository.findById(response.memberId)
    val newEventSendHistory = emailSendHistoryRepository.findById(1L)

    assertThat(newMember).isNotNull
    assertThat(newEventSendHistory).isNotNull
}

회원 가입 트랜잭션

2023-03-19T01:37:22.845+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MemberService.register]
2023-03-19T01:37:22.852+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
Hibernate: insert into member (email, nickname) values (?, ?)
2023-03-19T01:37:22.936+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T01:37:22.937+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MemberService.register]

축하 메일 발송 트랜잭션(예외 발생 및 롤백)

2023-03-19T01:37:22.959+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail]
2023-03-19T01:37:22.962+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
Hibernate: insert into email_send_history (message, send_at, target_id, type) values (?, ?, ?, ?)
2023-03-19T01:37:22.968+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T01:37:22.968+09:00 TRACE 27796 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail] after exception: java.lang.RuntimeException: send email exception
catch runtime exception

회원 가입 트랜잭션은 정상적으로 동작하여 쿼리가 발생하고 축하 메일 발송 트랜잭션은 오류가 나서 롤백하여 정상적으로 동작하는 것을 확인했습니다. 또한 이 방식은 트랜잭션을 commit 하기 때문에 db에 실제로 저장이 되는 로직임을 확인하고 이메일 발송 로직을 진행합니다. 즉 2, 4 번 문제를 해결했습니다.

다음은 전파 옵션을 변경하는 방법입니다.

@Transactional의 기본 전파 옵션은 REQUIRED입니다. 이는 새롭게 @Transactional 프록시 객체의 메서드를 호출할 때 기존에 트랜잭션이 생성되어 있으면 병합(merge)하는 설정입니다. REQUIRES_NEW는 새로운 트랜잭션을 생성하여 분리해주는 기능을 합니다. 트랜잭션의 전파전략에 대한 자세한 내용은 해당 글(@Transactional의 전파(propagation)와 격리(isolation)을 참조하시면 도움이 됩니다.

@Transactional(propagation = Propagation.REQUIRES_NEW)
fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String?) {
		...
    throw RuntimeException("send email exception")
}

설정을 바꿔주고 실행하면 정삭적으로 동작하는 것을 확인할 수 있습니다.

2023-03-19T01:43:05.226+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MemberService.registerProcess]
2023-03-19T01:43:05.238+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
Hibernate: insert into member (email, nickname) values (?, ?)
2023-03-19T01:43:05.334+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T01:43:05.338+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail]
2023-03-19T01:43:05.340+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
Hibernate: insert into email_send_history (message, send_at, target_id, type) values (?, ?, ?, ?)
2023-03-19T01:43:05.348+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T01:43:05.349+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail] after exception: java.lang.RuntimeException: send email exception
catch runtime exception
2023-03-19T01:43:05.359+09:00 TRACE 14776 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MemberService.registerProcess]

위의 퍼사드 방식과의 차이점이 있습니다. 퍼사드는 member 트랜잭션이 만들어지고 종료된 뒤에 mail 트랜잭션이 만들어지고 예외가 터져 롤백이 되었습니다. REQUIRES_NEW는 member 트랜잭션이 만들어지고 이후에 분리된 mail 트랜잭션이 만들어지고 예외와 롤백처리가 되고 분리되어 영향을 받지 않았던 member 트랜잭션이 종료가 됩니다.

이렇게 트랜잭션 전파 문제는 해결할 수 있었지만 여전히 기능은 강결합 상태이고 동기적인 방식입니다. 또한 하나의 트랜잭션 내부에서 분리하는 과정이라 db에 실제 저장되었는지 확인하지 못합니다. (해당 예제의 로직의 repository.save()는 persist 과정에서 db에 직접 저장하는 로직이 있습니다) 따라서 2번의 문제만을 해결한 방법입니다.

@Async

다음은 동기적인 방식의 성능 문제의 해결입니다. 회원가입 로직의 요청과 응답이 이메일 발송 기능 로직의 진행을 기다릴 필요가 없습니다. 따라서 기능적으로 분리할 필요가 있었습니다.

우리는 스프링의 @Async를 사용하여 비동기 방식으로 메일 인증 발송 로직을 처리할 것입니다.

@EnableAsync
@SpringBootApplication
class TransactionalEventListenerExampleApplication
@Async
fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String?) {
    ...
    throw RuntimeException("mail send exception")
}
fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

    // 1. 회원 등록
    val savedMember = register(registerMemberRequestData)

    // 2. 가입 축하 메일 발송 (메일 발송 기록은 db에 저장)
    mailService.sendSuccessRegisteredMemberMail(savedMember.id, savedMember.email)

    return RegisterMemberResponseData(memberId = savedMember.id!!)
}

어노테이션을 제외한 로직은 이전과 똑같습니다. 하지만 이번에는 try-catch로 예외를 잡아주지 않았습니다. 그런데도 테스트는 성공합니다.

2023-03-19T02:09:21.920+09:00 TRACE 13500 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MemberService.registerProcess]

비동기 로직에서는예외가 발생합니다.

2023-03-19T02:09:21.935+09:00 TRACE 13500 --- [         task-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail] after exception: java.lang.RuntimeException: mail send exception
2023-03-19T02:09:21.947+09:00 ERROR 13500 --- [         task-1] .a.i.SimpleAsyncUncaughtExceptionHandler : Unexpected exception occurred invoking async method: public void com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail(java.lang.Long,java.lang.String)

비동기 로직에서 발생한 예외는 회원가입 로직과 응답에 영향을 끼치지 않았습니다. 또한 트랜잭션도 오류가 나지 않았습니다. 이 이유에 대해서는 @Async의 동작방식과 Thread별 DB Connection 및 Connection별 트랜잭션에 대한 이해가 필요합니다.

@Async는 프록시 기반으로 작동하여 애노테이션에 지정한 executor 혹은 SimpleAsyncTaskExecutor를 사용하여 동작시킵니다. SimpleAsyncTaskExecutor는 각 작업마다 스레드를 만들고 비동기적으로 메서드를 실행시킵니다.

DataSourceTransactionManager 클래스를 보면 아래와 같은 내용이 기재되어있습니다.

Binds a JDBC Connection from the specified DataSource to the current thread, potentially allowing for one thread-bound Connection per DataSource.

하나의 스레드는 하나의 데이터 소스를 사용할 수 있고, 여러 커넥션을 사용할 수 있습니다. 또한 트랜잭션을 처리하기 위해서는 hikari cp 등의 커넥션 풀에서 커넥션을 가져와서 sql statement를 만들어주는 기능이 필요하고 commit 혹은 rollback 처리가 되면 해당 커넥션을 커넥션 풀에 반환합니다. 즉 하나의 트랜잭션을 처리하기 위해서는 하나의 커넥션이 필요한 것입니다.

@Async는 스레드를 새로 생성해서 로직을 실행시키므로 새로운 스레드에서 새로운 db connection을 가져와 새로운 트랜잭션을 생성합니다. 따라서 트랜잭션도 분리가 됩니다. 위에서 고민한 2,3 번 문제를 해결할 수 있었습니다. 다만 트랜잭션이 끝나면서 db 무결성 문제가 발생하게 되면 회원가입 로직은 롤백이 되는데 이미 메일이 발송되어버려 4번 문제는 해결할 수 없습니다.

기존 로직에서 더티체킹시 발생하는 db 무결성 오류를 발생시키고 메일 로직에 예외가 발생하지 않도록 변경해보겠습니다.

fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

    // 1. 회원 등록
    val savedMember = register(registerMemberRequestData)

    // error: db 에러
    savedMember.updateNickname("123456789012345678901234567890")

    // 2. 가입 축하 메일 발송 (메일 발송 기록은 db에 저장)
    mailService.sendSuccessRegisteredMemberMail(savedMember.id, savedMember.email)

    return RegisterMemberResponseData(memberId = savedMember.id!!)
}

더티 체킹시 에러 발생했지만

org.springframework.dao.DataIntegrityViolationException: could not execute statement; SQL [n/a]공

메일 발송은 성공합니다.

send mail success

...

2023-03-19T02:37:41.280+09:00 TRACE 24384 --- [         task-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail]

사용자는 회원가입은 실패했는데 메일을 받는 경험을 하게됩니다. 따라서 4번 문제의 해결이 필요합니다.

추가적으로 @Async를 사용한 비동기 방식은 예외처리를 다른 방식으로 해결해야합니다.

Effective Advice on Spring Async: Part 1 - DZone

Event

다음 방법은 Event입니다. 앞서 다양한 방법으로 여러 문제를 해결했지만 우리의 코드는 여전히 강한 결합도를 가지고 있습니다. 클라이언트는 회원가입을 하는 로직을 호출했지만 우리는 그 후속행위인 이메일을 발송하는 로직을 알고 있어야 합니다. 하나의 로직안의 두 가지의 기능이 결합되어있다고 볼 수 있습니다.

2022년 우아콘의 ‘회원시스템 이벤트기반 아키텍처 구축하기’에서는 이벤트 기반 아키텍처에 대해 설명을 했습니다. MSA 아키텍처에서 추구하는 느슨한 결합을 위해 이벤트 방식을 도입한 방법을 설명하고 있습니다. 이 영상의 내용을 통해 이벤트를 어떻게 정의해야하는지를 알아보겠습니다.

코드간의 강한 결합

  • 코드간의 결합이 일어나면 도메인을 다른 시스템으로 분리할 때 문제가 발생할 수 있습니다.
  • 시스템 분리를 하게 되면 기존 기능의 구현체를 http 통신으로 요청하는 로직으로 변경해야합니다. 하지만 이 또한 해당 도메인 외부의 후속행위를 알고 있으므로 강한결합이라고 할 수 있습니다.

비동기 방식에서도 강한 결합 발생

  • @Async를 사용해서 직접적인 의존을 제거할 수 있습니다.
  • 하지만 스레드의 의존을 제거할 뿐 후속행위의 의도는 그대로 알고 있으면 강한 결합입니다.

이벤트 방식에서의 강한 결합

  • 메시징 시스템을 사용한다고 느슨한 결합과 이벤트 아키텍처가 아닙니다. 발행한 메시지가 대상 도메인에게 기대하는 목적이 있으면 비동기 로직일 뿐입니다.
    • eventPublisher.sendSuccessMail(memberId)
    • 회원가입 → 이메일 발송 이벤트 발행 (후속행위의 의도를 담고 있음)
  • 회원가입 로직을 마쳤을 때, 회원가입이 완료되었다는 이벤트를 발행해야합니다. 메일 서비스는 회원가입 완료 이벤트를 구독하여 해당 이벤트가 발생했을 때 실행합니다. 회원가입 로직은 메일 서비스에 관여를 하지 않았습니다. 이로인해 두 시스템의 의존성이 느슨해집니다.
    • eventPublisher.registeredMember(memberID)
    • 회원가입 → 회원가입이 되었다는 이벤트 발행 (후속 행위의 의도를 담고있지 않음)

우리가 발행해야할 이벤트는 이벤트로 인해 달성하려는 목적이 아닌 도메인 이벤트 그 자체입니다. 우리는 회원가입 축하 메일 발송 이벤트를 발행하는 것이 아니라 회원가입이 되었다는 도메인의 상태 변경에 대한 이벤트를 발행하는 로직을 구현해야합니다.

회원시스템 이벤트기반 아키텍처 구축하기 #우아콘2022 #Day2_음식그이상의것을문앞으로

회원 등록을 할 때 발행할 이벤트, 이벤트 생성 로직, 이벤트 핸들러를 추가해줍니다. 이벤트 로직은 스프링에서 지원하는 ApplicationEvent와 이를 발행하는 ApplicationEventPublisher, 이벤트를 감지하고 핸들링하는 ApplicationEventListener를 스프링 4.2부터 제공하는 @EventListner를 통해 구현해보겠습니다.

fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

    // 1. 회원 등록
    val savedMember = register(registerMemberRequestData)

    // 2. 회원 등록 이벤트 발행
    val registeredMemberEvent = RegisteredMemberEvent(savedMember.id!!, savedMember.email!!)
    applicationEventPublisher.publishEvent(registeredMemberEvent)

    return RegisterMemberResponseData(memberId = savedMember.id)
}
data class RegisteredMemberEvent(
    val memberId: Long,
    val email: String
)
@Component
class RegisteredMemberEventHandler(
    private val mailService: MailService
) {

    @EventListener
    fun sendSuccessMail(registeredMemberEvent: RegisteredMemberEvent) {
        mailService.sendSuccessRegisteredMemberMail(registeredMemberEvent.memberId, registeredMemberEvent.email)
    }
}

정상적으로 로직은 실행됩니다. 하지만 코드상의 의존성이 제거되었을 뿐 여전히 동기적으로 동작합니다. 따라서 메일 서비스 안에서 예외가 발생했을 때 예외가 전파되기 때문에 이를 잡아주기 위해서 try-catch 를 해줘야합니다.

fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

    // 1. 회원 등록
    val savedMember = register(registerMemberRequestData)

    // 2. 회원 등록 이벤트 발행
    val registeredMemberEvent = RegisteredMemberEvent(savedMember.id!!, savedMember.email!!)

    try {
        applicationEventPublisher.publishEvent(registeredMemberEvent)
    } catch (e: RuntimeException) {
        println("catch exception")
    }

    return RegisterMemberResponseData(memberId = savedMember.id)
}

하지만 트랜잭션이 병합되어 rollback mark가 묻어 오류가 발생하기 때문에 이 또한 트랜잭션 분리를 해줘야 합니다.

@Transactional(propagation = Propagation.REQUIRES_NEW)
fun sendSuccessRegisteredMemberMail(memberId: Long?, emailAddress: String?) {
    val successRegisteredMember = SuccessRegisteredMemberMessageGenerator.generate(memberId!!)
    emailSender.send(successRegisteredMember, emailAddress)

    val emailSendHistory = EmailSendHistory(
        sendAt = LocalDateTime.now(),
        message = successRegisteredMember.message,
        targetId = memberId,
        type = EmailType.MEMBER_REGISTER_SUCESS
    )
    emailSendHistoryRepository.save(emailSendHistory)
    throw RuntimeException("send mail exception")
}

에러를 방지할 수 있지만 와닿는 방식은 아니라고 생각이 들것입니다. 코드간의 의존성을 해결했지만 여전히 하나의 트랜잭션, 동기적인 로직으로 결합도가 강합니다. 따라서 @Async를 추가해서 스레드를 분리하여 비동기로 처리하도록 구현하면 스레드가 분리되어 트랜잭션도 분리가 되고 회원 서비스에서 다른 로직의 수행을 기다리지 않아도 클라이언트에게 응답을 줄 수 있게 되었습니다.

@Component
class RegisteredMemberEventHandler(
    private val mailService: MailService
) {

    @EventListener
    @Async
    fun sendSuccessMail(registeredMemberEvent: RegisteredMemberEvent) {
        mailService.sendSuccessRegisteredMemberMail(registeredMemberEvent.memberId, registeredMemberEvent.email)
    }
}
@Test
fun `case 1 - 기본 예제`() {
    val memberRequestData = RegisterMemberRequestData(nickname = "ep", email = "ep@email.com")
    assertThrows<RuntimeException> {
        val response = memberService.registerProcess(memberRequestData)
        assertThat(response).isNotNull
        assertThat(response.memberId).isEqualTo(1L)
    }
}

메일 스레드에서 예외는 발생했지만 회원가입 로직은 성공적으로 동작하는 것을 확인할 수 있었습니다. 하지만 db 데이터 문제에서는 어떻게 동작할까요?

fun registerProcess(registerMemberRequestData: RegisterMemberRequestData): RegisterMemberResponseData {

    // 1. 회원 등록
    val savedMember = register(registerMemberRequestData)

    // error: db 에러
    savedMember.updateNickname("123456789012345678901234567890")

    // 2. 회원 등록 이벤트 발행
    val registeredMemberEvent = RegisteredMemberEvent(savedMember.id!!, savedMember.email!!)
    applicationEventPublisher.publishEvent(registeredMemberEvent)

    return RegisterMemberResponseData(memberId = savedMember.id)
}

역시 비동기로 동작하기에 DataIntegrityViolationException 에러는 발생했지만 메일은 발송이 되었습니다. 우리는 앞선 트랜잭션이 성공적으로 마무리가 된 것을 확인하고 메일을 발송하고 싶습니다. 이 문제를 해결하기 위해서는 트랜잭션 동기화 매니저(TransactionSynchronizationManager)를 이용하면 됩니다. 우선 @Async를 제거하고 트랜잭션 동기화 매니저의 로직을 추가하겠습니다. TransactionSynchronizationManager.registerSynchronization()를 사용하고 afterCommit를 오버라이딩하면 commit 이후의 동작을 정의할 수 있습니다. 

@EventListener
fun sendSuccessMail(registeredMemberEvent: RegisteredMemberEvent) {
    handleAsynchronously(registeredMemberEvent)
}

private fun handleAsynchronously(event: RegisteredMemberEvent) {
    if (TransactionSynchronizationManager.isActualTransactionActive()) {
        println("after commit")
        processAfterCommit(event)
    } else {
        println("now")
        processNow(event)
    }
}

private fun processNow(event: RegisteredMemberEvent) {
    mailService.sendSuccessRegisteredMemberMail(event.memberId, event.email)
}

private fun processAfterCommit(event: RegisteredMemberEvent) {
    TransactionSynchronizationManager.registerSynchronization(object : TransactionSynchronizationAdapter() {
        override fun afterCommit() {
            mailService.sendSuccessRegisteredMemberMail(event.memberId, event.email)
        }
    })
}

조금 코드가 복잡합니다. 실행결과는 다음과 같습니다.

2023-03-19T17:05:42.593+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MemberService.registerProcess]
2023-03-19T17:05:42.605+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
Hibernate: insert into member (email, nickname) values (?, ?)
2023-03-19T17:05:42.745+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
after commit
2023-03-19T17:05:42.747+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MemberService.registerProcess]
2023-03-19T17:05:42.775+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail]
mail send success
2023-03-19T17:05:42.778+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T17:05:42.780+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
2023-03-19T17:05:42.780+09:00 TRACE 19604 --- [           main] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.ep.transactional_event_listener_example.service.MailService.sendSuccessRegisteredMemberMail]
  1. member의 트랜잭션을 가져옵니다.
  2. member 로직을 진행하다가 event 로직을 발행합니다.
  3. 이벤트 리스너는 이벤트를 수신하고 로직을 실행시킵니다.
  4. TransactionSynchronizationManager.isActualTransactionActive() 로직에서 트랜잭션이 아직 활성화 상태이면 끝나기를 기다립니다.
  5. 활성화된 트랜잭션이 끝나면 새로운 트랜잭션으로 로직을 실행시킵니다.

따라서 DataIntegrityViolationException 이 발생하면 commit이 정상적으로 수행되지 않았으므로 이메일 발송되는 로직이 실행되지 않습니다. 이제 모든 문제가 해결된 것 같습니다. 하지만 이 로직은 사실 정상적으로 동작하지 않습니다. 이전에는 없던 새로운 문제가 생겼습니다. 이 내용의 설명을 위해 @Async를 제외하고 코드를 구현했는데 다음 내용을 보시면서 다시 확인하겠습니다.

ApplicationEventPublisher 기반으로 강결합 및 트랜잭션 문제 해결 - Yun Blog | 기술 블로그

[Spring] Spring의 @EventListener

Spring events and transactions — be cautious!

스프링 이벤트 기능을 사용할 때의 고려할 점

DTx - Junkman

@TrasanctionalEventLisner

위의 로직은 commit 이후에 동작을 하도록 설정할 수 있습니다. 스프링 4.2은 @EventListener 와 함께 Transactional afterCommit 로직을 간단한 어노테이션으로 구현할 수 있는 방법이 도입되었습니다.

@TransactionalEventListener는 이전에 작성한 TransactionSynchronizationManager 로직을 직접 구현할 필요없이 간단한 어노테이션만으로도 해결할 수 있습니다. 

@Component
class RegisteredMemberEventHandler(
    private val mailService: MailService
) {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    fun sendSuccessMail(registeredMemberEvent: RegisteredMemberEvent) {
        mailService.sendSuccessRegisteredMemberMail(registeredMemberEvent.memberId, registeredMemberEvent.email)
    }
}

이 방법은 트랜잭션을 분리해서 사용할 수 있습니다. 위의 로직과 동일하고 commit 이 일어난 뒤에 로직이 동작합니다. 하지만 이 방식은 기대하지 않는 결과를 만들어냅니다.

@Test
fun `case 1 - 기본 예제`() {
    val memberRequestData = RegisterMemberRequestData(nickname = "ep", email = "ep@email.com")
    val response = memberService.registerProcess(memberRequestData)
    assertThat(response).isNotNull
    assertThat(memberRepository.findById(1L).isPresent).isTrue
    assertThat(emailSendHistoryRepository.findById(1L).isPresent).isTrue // error
}

테스트 코드 마지막 줄에서 에러가 발생했습니다. emailSendHistory 데이터가 db에 저장되지 않는 것이었습니다.

TransactionSynchronization

afterCommitInvoked after transaction commit. Can perform further operations right after the main transaction has successfully committed.

void afterCommit()

Can e.g. commit further operations that are supposed to follow on a successful commit of the main transaction, like confirmation messages or emails.

NOTE: The transaction will have been committed already, but the transactional resources might still be active and accessible. As a consequence, any data access code triggered at this point will still "participate" in the original transaction, allowing to perform some cleanup (with no commit following anymore!), unless it explicitly declares that it needs to run in a separate transaction. Hence: Use PROPAGATION_REQUIRES_NEW for any transactional operation that is called from here.

Throws:[RuntimeException](http://docs.oracle.com/javase/8/docs/api/java/lang/RuntimeException.html?is-external=true) - in case of errors; will be propagated to the caller (note: do not throw TransactionException subclasses here!)

앞서 봤던 TransactionSynchronization 문서를 살펴보면 @TransactionEventListener를 사용하더라도 해당 트랜잭션을 가져온 스레드에서는 리소스가 여전히 활성화되어있고 액세스할 수 있다고 합니다. 별도의 트랜잭션 설정을 하지 않으면 기존 트랜잭션에 참여하는 개념이 됩니다. 하지만 앞선 트랜잭션은 이미 commit과 함께 종료가 되었고 재사용이 불가능한 트랜잭션입니다. 따라서 새 트랜잭션에서 실행되도록 선언해야지 정상적으로 트랜잭션을 완료할 수 있습니다.

이 문제를 해결하려면 이전에 우리가 사용했던 2가지의 방법을 사용하면 됩니다.

  • 트랜잭션을 시작하기만 하면 되는 @Transactional(propagation = Propagation.REQUIRES_NEW)을 사용할 수 있습니다.
  • 표준 @Transactional을 사용할 수 있지만 트랜잭션이 동일한 스레드 내에서 전파되기 때문에 새 트랜잭션을 실행해야 합니다. 메서드에 @Async어노테이션을 사용할 수 있습니다.

우리는 트랜잭션 분리와 함께 비동기 방식으로 로직을 처리하기 위해 @Async를 추가해줘서 문제를 해결할 수 있습니다. 동기적으로 처리해도 되는 경우에는 Transactional 전파 옵션으로도 분리가 가능합니다.

@Async
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
fun sendSuccessMail(registeredMemberEvent: RegisteredMemberEvent) {
    mailService.sendSuccessRegisteredMemberMail(registeredMemberEvent.memberId, registeredMemberEvent.email)
}
@Test
fun `case 1 - 기본 예제`() {
    val memberRequestData = RegisterMemberRequestData(nickname = "ep", email = "ep@email.com")
    val response = memberService.registerProcess(memberRequestData)
    assertThat(response).isNotNull
    assertThat(memberRepository.findById(1L).isPresent).isTrue
    assertThat(emailSendHistoryRepository.findById(1L).isPresent).isTrue
}

https://blog.pragmatists.com/spring-events-and-transactions-be-cautious-bdb64cb49a95

Conclusion

우리는 처음 로직에서 4가지 문제를 마주했고 @TransactionalEventListener를 통해 해결했습니다.

  1. 서비스와 서비스간의 강결합 → Event 로직과 Handler로 느슨한 결합
  2. 트랜잭션 결합 → @Async로 별도의 스레드(별도의 커넥션 or 트랜잭션)으로 트랜잭션 분리 or @Transactional을 통한 전파 옵션 설정으로 트랜잭션 분리
  3. 비효율적인 동기적 로직 → @Async를 통해 별도의 스레드로 작업하여 기능 분리. 빠른 응답
  4. 트랜잭션이 커밋된 후에 로직 후처리 → TransactionSynchronizationManager을 통해 after commit 로직 구현

이번 글에서는 이벤트 기반 로직의 쉬운 설명을 위해 Domain Event를 다루는 방법과 이벤트 스토어를 활용한 방식, 메시지큐를 사용한 방식, 최소 1회 보장(at least once), 이벤트 방식 고려사항 등의 이야기는 다루지 않았습니다. 안정적인 이벤트 기반의 아키텍처를 설계하기 위해서는 아직 넘어야할 산이 많습니다. 이야기가 궁금하신 분은 해당 키워드를 검색하거나 혹은 아래 참조 링크를 통해 확인해주시거나 다음 글을 기다려주시면 감사하겠습니다.

글에서 사용했던 예제 코드는 제 깃허브에서 확인하실 수 있습니다.

Reference

springframework:transaction:transactional_event_listener [권남]

[Spring] Spring의 @EventListener

Spring events and transactions — be cautious!

스프링 이벤트 기능을 사용할 때의 고려할 점

springframework:transaction:transactional_event_listener [권남]

분산 환경과 Event-Driven Architecture : SAGA , Event Sourcing, CQRS

At most once, at least once, exactly once

profile
EP
Hello!
이전 포스트

Spring Boot로 K6 & Grafana를 활용한 부하테스트 해보기

1개의 댓글

comment-user-thumbnail
공부는 혼자하는 거
2023년 11월 23일

진짜 좋은 글이네요 ㅎ

답글 달기