최근 ‘소프트웨어 설계’ 전공 과목에서 진행한 카풀 프로젝트를 리팩토링하며 도메인 주도 설계(DDD, Domain-Driven Design)의 헥사고날 아키텍처를 적용하였다. 따라서, 특정 도메인을 표현하기 위해 도메인 엔티티(Domain Entity)와 VO(Value Object) 객체를 정의하는 것을 중요하게 생각하게 되었다.

  또한, 기능을 하나씩 구현하면서 BDD Mockito를 기반으로 테스트 코드를 작성하며 스터빙(Stubbing)을 진행하였다.

@Test
@DisplayName("사용자 요약 정보 조회 성공")
void success() {
    // given
    User user = UserFixture.create();
    Long userId = user.getId().getValue();
    String nickname = user.getNickname().getValue();
    String email = user.getEmail().getValue();
    String studentNumber = user.getStudentInfo().getStudentNumber().getValue();
    
    GetUserSummaryQuery query = new GetUserSummaryQuery(userId);
    // ⭐️ 새로운 사용자 고유 번호 VO 객체(UserId)로 스터빙
    given(readUserPort.findByUserId(new UserId(userId))).willReturn(Optional.of(user));
    
    // when
    GetUserSummaryResult result = usecase.getUserSummary(query);
    
    // then
    assertEquals(nickname, result.nickname());
    assertEquals(email, result.email());
    assertEquals(studentNumber, result.studentNumber());
}

  사용자 고유 번호를 나타내는 UserId VO 객체의 생성자를 통해 새로운 객체로 만들어 스터빙한 코드에 적용하였다. 테스트를 실행해보면 아래 사진과 같은 에러가 발생하였다.

  에러 로그에서도 확인할 수 있듯이 스터빙 시 사용한 UserId 객체와 실제 테스트 시 메서드 호출에 사용된 UserId 객체가 서로 다른 객체이기 때문에 스터빙이 제대로 동작하지 않아 발생한 것이다.

  이러한 실패 상황을 보면서 “기존 테스트 코드에서 스터빙은 어떤 식으로 작성하였지?”라는 의문이 들어, 이전의 테스트 코드들을 확인해보았다.

@Test
@DisplayName("로그인 성공 시 AuthToken을 반환한다")
void success() {
    // given
    // ⭐️ any(...) 메서드를 사용한 스터빙
    given(readUserPort.findByUsername(any(Username.class))).willReturn(Optional.of(user));
    given(passwordPort.matches(any(RawPassword.class), any(EncodedPassword.class))).willReturn(true);
    given(authTokenIssuer.generateAuthToken(any(UserId.class))).willReturn(new AuthToken(accessToken, refreshToken));
    
    // when
    AuthToken authToken = usecase.signin(command);
    
    // then
    assertEquals(accessToken.getValue(), authToken.getAccessToken().getValue());
    assertEquals(refreshToken.getValue(), authToken.getRefreshToken().getValue());
    verify(refreshTokenStore, times(1)).save(any(UserId.class), any(RefreshToken.class));
}

  기존 테스트 코드들에서는 any(...) 메서드를 사용하여서 인자로 전달되는 값의 타입만 일치하면 스터빙이 진행되도록 설정하였었다.

  이전에도 “any(...)를 사용한 스터빙 시 주요 로직에 대한 검증이 확실한가?”는 의문이 든 적이 있다.

  당시에는 단위 테스트의 관심사는 테스트를 진행하는 메서드의 로직 검증이며, 메서드 호출 시 전달된 값이 별도의 변환 과정 없이 그대로 다음 로직으로 전달되는 경우가 많았기 때문에 any(...) 메서드를 사용한 스터빙에도 큰 문제를 느끼지 못하였다.

  그러나 도메인 엔티티와 VO를 중심으로 코드를 작성하고, 각 계층(모듈) 간 역할을 명확하게 분리하면서 상황은 달라졌다. 외부에서 내부로 전달된 DTO는 application 계층 내부에서 도메인 엔티티 혹은 VO로 변환되어 사용되었고, 반대로 외부 계층으로 데이터를 변환할 때 역시 DTO 변환 과정이 수행되었다.

  즉, 유즈케이스 내부에서는 단순히 전달받은 값을 그대로 사용하는 것이 아니라, 도메인으로써의 의미를 가지는 객체를 생성하게 되는 것도 핵심 비즈니스 로직에 포함하게 된 것이다.

  ‘타입’만 중요한 것이 아니라, 의도한 값 자체의 ‘의미’도 중요해졌다. 따라서 테스트 코드에서 any(...) 메서드를 사용하여 타입만 일치하면 스터빙이 동작하도록 하는 방식이 충분히 올바른 검증인가에 대한 의문을 가지게 되었다.

  특히, VO처럼 값 자체가 의미를 가지는 객체인 경우에는 잘못된 값으로 객체가 생성되거나 예상치 못한 DTO - VO 간 변환이 발생하더라도 any(...) 기반 스터빙은 이를 검증하지 못한다.

  이러한 과정을 거치며 다음과 같은 점들을 깨닫게 되었다.

• VO(Value Object)는 값 자체가 의미를 가지는 객체이므로, 올바른 비즈니스 동작과 정확한 테스트 검증을 위한 동등성 보장이 필요하다.
• any(...) 기반 스터빙은 객체 타입만을 기준으로 동작하기 때문에, DTO ↔ VO 변환 과정에서 발생할 수 있는 오류를 검증하지 못하는 한계가 존재한다.

  이러한 2가지 관점을 기반으로 “도메인 주도 설계에서 동등성 보장이 왜 중요한가?”는 질문의 답을 찾아나갈 수 있었다.

도메인 엔티티와 VO의 비교

도메인 엔티티(Entity)

  도메인 엔티티는 고유한 식별자(id, identifier)를 가지며, 특정 도메인에서 해결해야하는 대상을 표현한 객체이다.

  예를 들어 사용자, 주문, 게시글과 같은 객체들은 단순한 데이터의 묶음이 아니라 객체 자체가 특정 도메인을 표현한다.

User(id = 1, nickname = "test01")

User(id = 1, nickname = "test02")

  위 두 객체는 객체의 닉네임이 다르지만, 식별자(id)가 동일하기 때문에 같은 사용자를 의미한다.

  즉, 엔티티는 식별자(id)를 통해서 객체를 식별한다.

VO(Value Object)

  VO(Value Object)는 도메인 엔티티에서 사용되는 개별 요소의 의미를 감싸는 객체를 의미한다.

  예를 들어 이메일, 주소, 금액, 학번과 같은 객체들은 별도의 식별자를 가지는 것이 아니라 객체 자체의 값이 식별에 사용된다.

new Email("kim@gmail.com");

new Email("kim@gmail.com");

  위 두 객체는 서로 다른 인스턴스이지만, 이메일 값을 표현하는 VO이므로 내용(이메일 주소)이 같으므로 같은 객체로 취급되어야 한다.

  즉, VO는 객체의 주소(고유식별자, identity)가 아니라 내부 값(value)을 기준으로 비교되어야 한다.

동일성(identity)과 동등성(equality)

  객체를 비교할 때 동일성(identity)와 동등성(equality)이란 개념이 사용된다.

  동일성(identity)이란 두 객체가 완전히 같은 지를 의미하며, Java에서는 참조 타입인 두 객체의 주소가 같은 경우를 의미한다.

  동등성(equality)이란 두 객체가 동일한 정보를 가지고 있다는 것을 의미하며, 두 변수가 저장된 주소가 다르더라도 가지는 값이 같으면 같은 객체를 의미하는 경우를 일컫는다.

VO에서 동등성이 보장되어야 하는 이유

  동등성의 의미에서 VO에서 동등성이 보장되어야하는 이유를 알 수 있다.

  VO는 내부 값을 특정한 의미를 담아 감싸는 래핑된 객체를 의미하며, 가지는 값 자체가 해당 객체의 정체성이라 할 수 있다. 즉, 해당 객체가 실질적으로 의미하는 값이 같으면 같은 객체로 취급해야한다.

  해당 개념은 동등성을 의미하며, VO에서 동등성이 보장되어야 하는 이유이다.

동등성이 필요한 곳

  객체의 동등성이 보장되어야지만 동작에 이상이 없는 것들은 다음과 같다.

• VO의 개념
• Mockito Stubbing
• Collection - HashSet

  위 개념들에서 동등성(equality)이 보장되어야하는 이유를 찾을 수 있다. VO의 개념에서는 앞서 설명한 것과 같은 내용이다.

  Mockito Stubbing은 해당 포스팅의 서론 부분에서 언급한 부분과 같이, 스터빙된 메서드의 인자가 동등한 경우에만 스터빙에서 설정한 동작대로 테스트를 진행할 수 있다.

  물론 동등성을 통한 비교외에도 argThat(...) 메서드를 통해서 스터빙을 진행할 수도 있다.

@Test
@DisplayName("사용자 요약 정보 조회 성공")
void success() throws Exception {
    // given
    String nickname = UserFixture.nickname();
    String email = UserFixture.email();
    String studentNumber = UserFixture.studentNumber();
    
    GetUserSummaryResult result = new GetUserSummaryResult(nickname, email, studentNumber);
    // ⭐️ argThat(...) 메서드를 통한 스터빙
    given(getUserSummaryUsecase.getUserSummary(argThat(arg -> arg.userId() == 1L))).willReturn(result);
    
    // when
    ResultActions resultActions = mockMvc.perform(
            get("/api/v1/users/summary")
                    .with(user(userDetails))
    ).andDo(print());
    
    // then
    resultActions
            .andExpect(status().isOk())
            .andExpect(jsonPath("$.data.nickname").value(nickname))
            .andExpect(jsonPath("$.data.email").value(email))
            .andExpect(jsonPath("$.data.studentNumber").value(studentNumber));
}

  이 외에도 HashSet과 같은 Collection 사용 시에도 동등성이 보장되어야 한다. 정확하게는 클래스에 hashCode()와 equals() 메서드를 구현해야한다.

equals()와 hashCode()

equals()

  Java에서는 객체의 동등성(equality)를 비교하기 위해서는 Object 클래스에서 제공하는 equals(Object o) 메서드를 오버라이딩 해야한다.

  equals() 메서드는 기본적으로 객체의 주소를 기반으로 비교를 수행하기 때문에 내부 값이 동일한 객체더라도 서로 다른 인스턴스로 생성된 경우에는 다른 객체로 판별한다.

  그러나 VO는 객체 내부 값 자체가 객체의 의미를 결정한다. 따라서, 서로 다른 인스턴스라 하더라도 객체가 가지는 값이 같다면 같은 객체로 판별하여야 한다.

  따라서, VO에서는 값 기반 비교가 가능하도록 equals() 메서드를 오버라이딩하여 값을 비교하도록 구현해야한다.

hashCode()

  hashCode는 equals()와 동시에 자주 언급되는 메서드이다. 또한, equals()를 구현할 경우 무조건 hashCode()도 구현해야한다고 언급된다.

  equals()가 객체의 동등성을 보장하는데 굳이 hashCode()도 구현해야하는 이유가 뭘까?

  그 이유는 Java에서는 Hash 기반 Collection(HashSet, HashMap, HashTable)이 존재하기 때문이다. 해당 컬렉션들이 객체를 저장하고 조회할 때 hashCode()와 equals()를 같이 사용한다.

  HashSet과 같은 컬렉션은 객체를 저장할 때, 우선 hashCode()를 기반으로 버켓(Hash Bucket)의 위치를 찾는다. 이후 같은 해시 값을 가지는 객체들에 대해서는 equals() 메서드를 통해 동일한 객체인지 판별한다. HashSet의 경우에는 Set(집합)이기 때문에 동일한 객체가 중복으로 저장되더라도 하나의 객체만 저장된다.

  따라서, equals()를 구현하여 두 객체의 동등성을 보장하였더라도 hashCode()를 구현하지 않는다면 해당 객체를 Hash Collection에 저장할 때 중복 저장되는 등 예기치 못한 동작으로 이어질 위험성이 존재하기 때문에 equals()와 hashCode()를 동시에 구현하여야 한다.

마무리

  헥사고날 아키텍처를 기반으로 도메인 주도 설계를 심도있게 공부하고 있다. 애그리거트, 바운디드 컨텍스트와 같이 실제 비즈니스 규모에서 사용되는 거시적인 개념들을 이해하고 사용하는 것에는 한계가 있어 우선은 도메인의 개념과 의미를 중점으로 구조를 고민하고 있다.

  사실 기존에는 equals()와 hashCode()를 단순히 “같은 객체인지를 판별하기 위한 메서드” 정도로만 이해하고 있었다. 또한 VO(Value Object)를 적극적으로 사용하지 않았기 때문에, 객체 비교 역시 대부분 엔티티의 식별자(id)를 기준으로 수행하였고 실질적으로 객체의 동등성을 깊게 고민해본 경험이 많지 않은 것 같다.

  그러나 카풀 프로젝트를 리팩토링하며 계층(모듈) 간 경계를 명확히 분리하고, DTO ↔ Entity, VO 변환 규칙을 엄격하게 적용하는 구조로 설계를 변경하면서 객체 동등성의 중요성을 체감할 수 있었다.

  이전에는 단순히 “객체 비교에는 equals()를 사용한다” 정도로만 이해하고 있었다면, 이번 경험을 통해 VO는 값 자체가 의미를 가지는 객체이기 때문에 객체의 주소가 아니라 값 기반으로 같음을 판별할 수 있어야 하며, 이를 위해 Java에서는 Object.equals()와 hashCode()를 오버라이딩하여 동등성을 보장해야한다는 점을 실제적인 경험을 통해 이해할 수 있었다.

  트랜잭션 아웃박스 패턴을 구현하며, 발행자(Publisher)에서 발행된 이벤트는 외부 메시지 브로커(RabbitMQ)로 전달되어야 한다.

  필자는 이벤트 발행을 2가지 단계로 나누어서 처리하였다.

1. 트랜잭션 커밋 이후 즉시 발행 (빠른 이벤트 발행 및 적절한 폴링 주기 사용 목적)
2. 이벤트 아웃박스 폴링을 통한 재발행 (발행 보장)

  이전 포스팅(Transactional Outbox Pattern 도입기 3 - 메시지 브로커 및 이벤트 외부 발행 구조 도입)에서는 트랜잭션 커밋 이후 즉시 이벤트를 발행하는 로직에 대해 설명하였다.

  이번 포스팅에서는 Chris Richardson의 Microservice Architecture - Pattern: Transactional outbox에서 소개하는 Message Relay 개념을 기반으로, 아웃박스를 주기적으로 폴링하여 이벤트를 재발행하는 구조를 적용한 과정을 공유하고자 한다.

이벤트 아웃박스 폴링의 목적

  이벤트 아웃박스를 폴링하는 목적은 이벤트 발행 보장이다.

  외부 메시지 브로커를 도입하면서 Publisher/Consumer를 논리적·물리적으로 분리하였다. 이를 통해 책임 분리와 장애 격리와 같은 이점을 얻었지만, 이벤트가 메시지 브로커로 정상적으로 발행되어야만 이후 처리가 가능하다.

  트랜잭션 아웃박스 패턴의 궁극적인 목표 역시 이벤트 처리 실행을 보장하는 데 있다.

  이벤트를 아웃박스로 변환해 저장한 이유 또한, 이벤트를 아웃박스 형태로 영속화하여 메시지 브로커로 발행이 실패하더라도 재발행을 통해 실행을 보장하기 위함이다.

  Chris Richardson이 소개한 트랜잭션 아웃박스 패턴에서는 메시지 릴레이(Message Relay)라는 이름으로 소개된다.

  본 프로젝트에서는 비밀번호 초기화 및 이메일 인증 코드 발송 요청 기능에 이벤트 기반 구조를 도입하여 결합도를 낮추었다. 또한, 실행 보장을 위해 트랜잭션 아웃박스 패턴을 적용하였다.

  도메인 이벤트를 아웃박스의 형태로 저장하고 이를 폴링하는 구조이므로, 해당 컴포넌트(클래스)를 이벤트 폴러(Event Poller)라고 명명하였다.

폴링 대상 이벤트

  발행자(Publisher)는 이벤트 아웃박스를 조회해 발행해야 한다. 발행 대상 이벤트는 다음과 같다.

• 아직 발행되지 않은 이벤트
• 발행에 실패한 이벤트

  각 이벤트 아웃박스들의 상태는 WAITING 또는 FAILED이다.

  또한, 무분별한 재발행을 방지하기 위해 상태 뿐만 아니라 생성 시간(createdAt), 실패 횟수(failCount) 등을 통해서 조건에 맞는 이벤트만 조회해야한다.

1) 아직 발행되지 않은 이벤트

  초기 이벤트 아웃박스 생성 시 상태는 발행대기(WAITING) 상태이다.

  트랜잭션 커밋 이후 즉시 발행을 수행하지만, 예기치 못한 이유로 이벤트 외부 발행 리스너가 동작하지 않는 경우 이벤트는 여전히 WAITING 상태로 남을 수 있게 된다. 따라서, 즉시 발행되지 않은 이벤트를 폴링해서 재발행해야 한다.

  다만, 아웃박스 생성과 즉시 발행 사이의 짧은 시간 동안 폴링이 수행되면 동일 이벤트가 중복 발행될 수 있기 때문에 이를 방지하기 위해 30초가 지난 아웃박스만 조회하도록 하였다.

  또한, 이벤트에는 유효 시간이 존재한다. 예를 들어 비밀번호 초기화 코드는 3분의 유효 시간을 가지며, 해당 시간이 지난 후 발행되는 이벤트는 의미가 없다.

  이벤트 조회 시 도메인 이벤트별로 개별 유효 시간을 고려하면 쿼리가 복잡해지는 문제가 있다. 도메인 이벤트의 종류가 늘어날수록 조회 쿼리가 늘어난다는 문제도 존재한다.

  따라서, 가장 긴 유효 시간을 기준으로 조회 범위를 제한하고, 개별 이벤트의 유효성 검증은 소비자 측에서 처리하도록 설계하였다.

  현재 비밀번호 초기화 및 이메일 인증 이벤트의 유효 시간은 모두 3분이므로, 생성 후 3분을 초과한 이벤트는 조회 대상에서 제외된다.

@Service
@RequiredArgsConstructor
public class EventOutboxService {
    private final EventOutboxRepository eventOutboxRepository;
    
    // ...
    private final int WAITING_EVENT_MIN_AGE_SECONDS = 30;
    private final int WAITING_EVENT_MAX_AGE_MINUTES = 3;
    
    public List<EventOutbox> readAllPollingEventOutbox() {
        List<EventOutbox> waitingEventOutboxes = eventOutboxRepository.findByStatusAndCreatedAt(
                EventOutboxStatus.WAITING,
                LocalDateTime.now().minusMinutes(WAITING_EVENT_MAX_AGE_MINUTES),
                LocalDateTime.now().minusSeconds(WAITING_EVENT_MIN_AGE_SECONDS)
        );
    }

    // ...
}

  따라서, 폴링 시 이벤트 아웃박스를 조회할 때 생성 후 30초 경과, 3분 이내 이벤트만 조회하도록 한다.

개선사항. 이벤트 아웃박스에 expired_at 컬럼을 추가

  현재 구현에서는 각 도메인 이벤트의 유효시간을 기준으로 EventOutboxService의 WAITING_EVENT_MAX_AGE_MINUTES 값을 설정해야 한다. 현재는 모든 도메인 이벤트의 유효 시간이 3분으로 동일하지만, 서로 다른 유효 시간을 가진 이벤트가 추가된 경우 해당 값을 직접 수정해야한다. 이는 도메인 이벤트에 암묵적으로 의존하는 결합으로 이어진다.

  또한, 상대적으로 유효 시간이 짧은 이벤트는 다른 이벤트의 유효 시간으로 인해 일단 조회되어 소비자에서 이를 필터링하게 된다. 그러나 이 방식의 문제점은 결국 유효 시간이 지난 WAITING 이벤트도 발행하게 되어 불필요한 자원 낭비로 이어질 수 있다.

  이러한 문제점을 해결하기 위해 이벤트 아웃박스에 expired_at 컬럼을 추가할 수 있다. 이를 통해 현재 시간을 기준으로 유효 시간이 지난 이벤트는 조회 대상에서 제외되도록 할 수 있다.

@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Entity(name = "event_outbox")
public class EventOutbox {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "event_id", nullable = false, length = 26)
    private String eventId;
    
    @Column(name = "event_type", nullable = false)
    private String eventType;
    
    @Column(nullable = false)
    private String payload;
    
    @Column(nullable = false)
    @Enumerated(EnumType.STRING)
    private EventOutboxStatus status;
    
    @Column(name = "created_at", nullable = false)
    private LocalDateTime createdAt;
    
    @Column(name = "last_retried_at")
    private LocalDateTime lastRetriedAt;

    // ⭐️ expired_at 컬럼 추가
    @Column(name = "expired_at")
    private LocalDateTime expiredAt;
    
    @Column(name = "fail_count")
    private int failCount;

    // ...
}

이후, 이벤트 아웃박스 조회 시 현재 시간이 expired_at보다 지난 아웃박스 레코드들은 조회하지 않도록 한다.

2) 발행에 실패한 이벤트

  이벤트 발행을 시도했지만, 메시지 브로커 에러 등의 이유로 실패할 수 있다.

  발행에 실패한 이벤트의 상태는 발행 실패(FAILED)로, 유효 시간 내에 재시도를 진행해야한다.

  WAITING 상태의 아웃박스를 조회할 때 생성 시간을 통해 필터링하여 조회하였듯, FAILED 상태의 이벤트는 실패 횟수(failCount)를 통해 폴링 대상 아웃박스를 필터링한다.

발행 실패 이벤트 재발행 기준: 실패 횟수 vs 만료 기한

  본 프로젝트에서 트랜잭션 아웃박스 패턴을 적용한 대상은 유효 기간이 존재하는 이벤트이다.

  그러나 트랜잭션 아웃박스 패턴은 이메일 발송과 같은 부가 로직의 실행 보장 뿐만 아니라, MSA 환경에서 모듈 간 데이터 정합성을 유지하기 위한 방법으로도 사용된다. 이러한 경우에는 이벤트의 유효 기간이 존재하지 않으며, 데이터의 정합성이 궁극적으로 보장되어야한다는 책임만 존재한다.

  이벤트 발행 실패의 주요 원인은 메시지 브로커의 장애이다. 장애가 지속되는 상황에서는 재발행을 시도하더라도 동일하게 실패할 가능성이 높다. 또한, 메시지 자체가 잘못된 경우에도 반복적인 발행 실패가 발생한다. 따라서, 별도의 제한 없이 재발행을 시도할 경우, 실패할 이벤트를 지속적으로 발행하게 되어 불필요한 자원 낭비로 이어진다.

  이러한 문제를 방지하기 위해 실패 횟수 failCount를 기준으로 재발행 횟수를 제한할 수 있다. 일정 횟수 이상 실패한 이벤트는 더 이상 자동 재발행하지 않고, 메시지 브로커 복구 이후 별도의 배치 작업을 통해 수동으로 발행하거나, 다른 모듈과 데이터 동기화 작업을 별도로 실시하여 정합성을 맞추도록 할 수 있다.

  한편, failCount 대신 만료 기한(expiredAt)을 기준으로 실패한 이벤트를 필터링하는 방식도 고려할 수 있다. 그러나, FAILED 상태의 이벤트는 이미 발행에 실패한 이력이 있기 때문에 만료 기한 내 재시도를 하더라도 지속적으로 실패할 수 있다. 더불어 만료 기한이 긴 이벤트의 경우에는 해당 기한동안 지속적으로 재발행을 시도하게 되어 시스템 부하를 일으킬 수 있다.

  따라서, 현재 구현에서는 failCount를 기준으로 재발행 횟수를 제한하고, 3회 이상 실패한 이벤트는 더 이상 조회 대상에서 제외하도록 설계하였다.

@Service
@RequiredArgsConstructor
public class EventOutboxService {
    private final EventOutboxRepository eventOutboxRepository;
    
    private final int MAX_FAIL_COUNT = 3;
    private final int WAITING_EVENT_MIN_AGE_SECONDS = 30;
    private final int WAITING_EVENT_MAX_AGE_MINUTES = 3;
    
    public List<EventOutbox> readAllPollingEventOutbox() {
        // 발행 대기 중인 이벤트를 조회
        // 생성 후 30초 뒤, 생성 후 3분 이내 이벤트만 조회
        List<EventOutbox> waitingEventOutboxes = eventOutboxRepository.findByStatusAndCreatedAt(
                EventOutboxStatus.WAITING,
                LocalDateTime.now().minusMinutes(WAITING_EVENT_MAX_AGE_MINUTES),
                LocalDateTime.now().minusSeconds(WAITING_EVENT_MIN_AGE_SECONDS)
        );
        
        // 발행에 실패한 이벤트를 조회
        // 3회 미만 실패한 이벤트만 조회
        List<EventOutbox> failedEventOutboxes = eventOutboxRepository.findByStatusAndFailCount(
                EventOutboxStatus.FAILED,
                MAX_FAIL_COUNT
        );
        
        return Stream.concat(waitingEventOutboxes.stream(), failedEventOutboxes.stream()).collect(Collectors.toList());
    }

    // ...
}

  해당 도메인 서비스 클래스 EventOutboxService의 readAllPoolingEventOutbox() 메서드는 WAITING 상태의 이벤트와 FAILED 상태의 이벤트를 조회한 뒤, 이를 하나의 리스트로 병합하여 반환한다.

JSON 문자열 타입의 이벤트 발행

  위 코드는 RabbitMQ로 메시지를 발행하는 RabbitMQEventPublisher.publish(DomainEvent) 메서드이다.

  해당 메서드는 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) 리스너에 의해서 트랜잭션 커밋 이후 시점에서 전달되는 도메인 이벤트 객체를 발행하는데 사용된다.

  메서드 내부에서 RabbitTemplate.convertAndSend(...)를 호출하여 도메인 이벤트 객체를 JSON 형태로 직렬화한 뒤 메시지 브로커로 전송한다.

@Configuration
public class RabbitMQConfig {
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
    
    @Bean
    public MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

  RabbitMQ 관련 빈 등록 시, Jackson2JsonMessageConverter를 메시지 컨버터로 설정하였다.

  따라서, Java 객체를 전달할 경우 내부적으로 JSON 직렬화 과정을 거치며 RabbitTemplate.convertAndSend(...)를 사용하게 된다.

  그러나, 이벤트 아웃박스 폴링을 통해 발행하는 경우에는, 데이터베이스에 저장된 아웃박스를 기반으로 메시지를 생성해야한다.

  이벤트 아웃박스 엔티티(EventOutbox)의 payload 컬럼에는 도메인 이벤트가 JSON 문자열 형태로 저장되어있다. 따라서, RabbitMQEventPublisher.publish(DomainEvent)를 사용하려면, 해당 문자열을 다시 도메인 객체로 역직렬화하는 과정이 필요하다.

  이 과정을 불필요한 변환 비용을 발생시키므로, JSON 문자열을 그대로 발행하는 방식을 선택하였다. 이를 위해 RabbitTemplate.send(...)을 사용하면 역직렬화없이 문자열 기반 메시지를 직접 전송할 수 있다.

  따라서, 이벤트 아웃박스에 저장된 payload를 그대로 활용하여 메시지를 발행할 수 있도록, RabbitMQEventPublisher에 문자열 기반 발행 메서드를 추가하였다.

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMQEventPublisher implements DomainEventExternalPublisher {
    private final RabbitTemplate rabbitTemplate;
    private final RabbitMQPropertyResolver rabbitMQPropertyResolver;
    
    // ...
    
    /**
     * RabbitMQ 메시지브로커로 문자열 형태의 이벤트를 발행하는 메서드
     *
     * <p> {@code CorrelationData}를 통해 메시지브로커로의 발행 여부를 확인
     *
     * <p>
     *
     * @param eventId   이벤트 식별자(ULID)
     * @param eventType 도메인 이벤트 클래스 타입
     * @param payload   이벤트 객체를 JSON 형태로 변환한 문자열
     * @return          실행 결과 {@link EventPublishResult}를 감싸고 있는 {@code CompletableFuture} 객체
     */
    public CompletableFuture<EventPublishResult> publishRaw(String eventId, String eventType, String payload) {
        CorrelationData correlationData = new CorrelationData(eventId);
        
        CompletableFuture<EventPublishResult> result = correlationData.getFuture().thenApply(confirm -> {
            if (confirm.isAck()) {
                return EventPublishResult.success(eventId);
            } else {
                return EventPublishResult.fail(eventId, confirm.getReason());
            }
        }).exceptionally(ex -> EventPublishResult.fail(eventId, ex.getMessage()));
        
        try {
            Message message = MessageBuilder
                    .withBody(payload.getBytes(StandardCharsets.UTF_8))
                    .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                    .build();

            rabbitTemplate.send(
                    rabbitMQPropertyResolver.getPublishExchange(eventType),
                    rabbitMQPropertyResolver.getPublishRoutingKey(eventType),
                    message,
                    correlationData
            );
        } catch (Exception e) {
            result.complete(EventPublishResult.fail(eventId, e.getMessage()));
        }
        
        return result;
    }
}

  CorrelationData를 통한 Publish Confirm 결과 수신 및 EventPublishResult 객체 변환 로직은 기존과 동일하다.

  RabbitTemplate.convertAndSend(...) 대신 RabbitTemplate.send(...)를 사용하기 때문에 JSON 문자열 형태의 이벤트를 Message 객체로 생성하여 전달해야 한다.

  이때, payload를 바이트 배열 형태로 설정하고, Content-Type을 JSON으로 명시한다.

  이처럼 메시지의 콘텐츠 타입을 지정하면 Jackson2JsonMessageConverter를 사용하는 수신자(Consumer)에서 JSON 메시지를 Java 객체로 역직렬화할 수 있다.

이벤트 폴링 주기

  이벤트 폴러(Message Relay)를 통해서 주기적으로 이벤트 아웃박스를 조회하여, 미발행 이벤트와 발행에 실패한 이벤트를 재발행해야 한다.

  이때 폴링 주기는 시스템 성능에 직접적인 영향을 미친다. 폴링 주기가 길면 한 번에 많은 이벤트를 처리하게 되어 부하가 증가할 수 있고, 반대로 주기가 너무 짧으면 잦은 조회 쿼리로 인해 DBMS에 부담을 줄 수 있다.

  본 프로젝트에서는 Spring Event를 통해 트랜잭션 커밋 이후 즉시 이벤트를 발행하는 구조를 함께 사용하고 있다. 따라서, 대부분의 이벤트는 즉시 발행 단계에서 처리되며 폴러에 의해 조회되는 아웃박스의 수는 많지 않다. 이에 따라 이벤트 폴링 주기는 30초로 설정하였다.

  이것이 Spring Event를 통하여 컴시 완료 시점에 이벤트 발행 기능을 도입한 이유이기도 하다.

폴링한 이벤트 발행

  이제 실제로 조회(폴링)한 이벤트를 발행해야한다.

@Slf4j
@Service
@RequiredArgsConstructor
public class EventOutboxPoller {
    private final EventOutboxService eventOutboxService;
    private final DomainEventExternalPublisher eventExternalPublisher;
    private final Executor messagePublishTaskExecutor;
    
    /**
     * 주기적으로 이벤트 아웃박스를 조회하여, 재발행을 시도하는 폴링 메서드
     *
     * <p> 30초마다 미발행 및 발행 실패 이벤트 아웃박스를 조회하여 재발행 시도
     *
     * <p> 미발행 이벤트의 경우에는 초기 발행 시점과 동시성 문제를 예방 및 이벤트 최대 유효 시간을 고려하여 하고자 생성 후 30초 후, 3분 이내의 이벤트만을 조회
     *
     * <p> 재발행 후 상태 업데이트
     */
    @Scheduled(fixedRate = 30, timeUnit = TimeUnit.SECONDS)
    public void poll() {
        List<EventOutbox> eventOutboxes = eventOutboxService.readAllPollingEventOutbox();
        
        // 조회한 이벤트 아웃박스들을 발행
        List<CompletableFuture<EventPublishResult>> results = eventOutboxes.stream()
                .map(eventOutbox -> CompletableFuture.supplyAsync(
                                eventPublishTask(eventOutbox),
                                messagePublishTaskExecutor
                        )
                        .thenCompose(future -> future)
                        .exceptionally(ex -> {
                            log.error("Failed to published event \"{}\"", eventOutbox.getEventId(), ex);
                            return EventPublishResult.fail(eventOutbox.getEventId(), ex.getMessage());
                        })
                )
                .toList();
        
        // 모든 이벤트 아웃박스에 대한 메시지 발행이 될 때까지 대기
        CompletableFuture.allOf(results.toArray(new CompletableFuture[0])).join();
        
        // 메시지 발행이 모두 완료되고 난 다음, 이벤트 아웃박스들의 상태를 업데이트
        updateStatus(results);
    }

    // ...
}

  이벤트 폴러(EventOutboxPoller)는 주기적으로 이벤트 아웃박스를 조회한 뒤, 각 이벤트에 대해 비동기적으로 발행 작업을 수행한다.

  이벤트 발행 시에는 이전 포스팅에서 정의한 MessagePublishTaskExecutor 쓰레드풀을 사용한다. 이를 통해 여러 이벤트를 병렬로 처리하여 발행 시 처리량을 높일 수 있다.

  쓰레드풀을 지정하여 이벤트 발행 작업을 실행하기 위해 CompletableFuture.supplyAsync(Supplier, Executor) 정적 메서드를 사용한다. 첫 번째 인자로 실행할 작업을 Suppllier로 제공하며, 작업을 실행할 쓰레드풀을 두 번째 인자로 지정한다.

  이때 각 이벤트 발행 작업은 별도의 비동기 작업으로 실행되며, 실행 결과는 CompletableFuture<EventPublishResult> 형태로 반환된다.

/**
 * 이벤트 아웃박스 발행 작업을 나타내는 Supplier 메서드
 *
 * @param eventOutbox 이벤트 아웃박스
 * @return 이벤트 메시지 발행 결과 CompletableFuture 객체를 감싸는 Supplier
 */
private Supplier<CompletableFuture<EventPublishResult>> eventPublishTask(EventOutbox eventOutbox) {
    return () -> eventExternalPublisher.publishRaw(
            eventOutbox.getEventId(), eventOutbox.getEventType(), eventOutbox.getPayload()
    );
}

  이벤트 발행 작업은 Supplier 함수형 인터페이스로 넘겨준다.

  앞서 정의한 DomainEventExternalPublisher의 구현체인 RabbitMQEventPublisher.publishRaw(...) 메서드를 실행할 작업으로 정의한다.

  .publishRaw(...)의 반환값은 CorrelationData의 결과값을 받아 비동기적으로 처리되는 CompletableFuture<EventPublishResult>이다. 이는 Supplier가 반환하는 실제 객체이다.

  한편 CompletableFuture.supplyAsync(...)는 이벤트 발행 실행 결과를 감싸는 CompletableFuture를 반환하므로, 전체 반환 타입은 CompletableFuture<CompletableFuture<EventPublishResult>>가 된다. 따라서, .thenCompose(future -> future)를 통해 중첩된 CompletableFuture를 제거한다.

  이 과정에서 예외가 발생하면 실패 상태의 EventPublishResult를 반환하도록 처리한다.

  최종적으로 각 이벤트의 발행 결과는 List<CompletableFuture<EventPublishResult>> results에 저장되며, 이후 결과에 따라 이벤트 아웃박스의 상태를 갱신한다.

// 모든 이벤트 아웃박스에 대한 메시지 발행이 될 때까지 대기
CompletableFuture.allOf(results.toArray(new CompletableFuture[0])).join();

  모든 이벤트 아웃박스에 대한 메시지 발행이 완료될 때까지 대기한다.

  이는 모든 이벤트가 발행이 완료되고 난 후 일괄적으로 상태 갱신 작업을 수행하기 위한 목적으로, CompletableFuture.allOf(CompletableFuture...).join()을 사용하여 모든 비동기 이벤트 발행 작업이 종료될 때까지 블로킹한다.

이벤트 상태 갱신

/**
 * 이벤트 메시지 발행 결과를 바탕으로 이벤트 아웃박스 상태{@code status}를 업데이트하는 메서드
 *
 * <p> 메서드 호출부에서 모든 이벤트 아웃박스에 대한 메시지 발행이 되기까지 대기하였으므로 결과 객체{@link EventPublishResult}를 즉시 반환
 *
 * <p> 대기 및 실패 이벤트 수를 고려하여 일괄 업데이트 수행
 *
 * @param results 이벤트 메시지 발행 결과 CompletableFuture 리스트
 */
private void updateStatus(List<CompletableFuture<EventPublishResult>> results) {
    List<String> publishedEventIds = new ArrayList<>();
    List<String> failedEventIds = new ArrayList<>();
    
    results.stream()
            .map(result -> result.join())
            .forEach(result -> {
                if (result.isSuccess()) {
                    publishedEventIds.add(result.eventId());
                } else {
                    failedEventIds.add(result.eventId());
                }
            });
    
    if (!publishedEventIds.isEmpty()) {
        eventOutboxService.updateToPublishedByEventIds(publishedEventIds);
    }
    if (!failedEventIds.isEmpty()) {
        eventOutboxService.updateToFailedByEventsIds(failedEventIds);
    }
}

  폴링을 통해 이벤트 발행이 완료되면, Publish Confirm 결과를 바탕으로 이벤트 아웃박스의 상태를 갱신해야 한다.

  이는 EventPoller.updateStatus(List<CompletableFuture<EventPublishResult>> results) 메서드에서 수행하게 된다.

  발행 결과 리스트를 순회하며 각 이벤트의 성공 여부에 따라 성공 이벤트 ID와 실패 이벤트 ID를 구분하여 저장한 뒤, 이를 기반으로 상태를 일괄 업데이트한다.

EventPoller 전체 코드

@Slf4j
@Service
@RequiredArgsConstructor
public class EventOutboxPoller {
    private final EventOutboxService eventOutboxService;
    private final DomainEventExternalPublisher eventExternalPublisher;
    private final Executor messagePublishTaskExecutor;
    
    /**
     * 주기적으로 이벤트 아웃박스를 조회하여, 재발행을 시도하는 폴링 메서드
     *
     * <p> 30초마다 미발행 및 발행 실패 이벤트 아웃박스를 조회하여 재발행 시도
     *
     * <p> 미발행 이벤트의 경우에는 초기 발행 시점과 동시성 문제를 예방 및 이벤트 최대 유효 시간을 고려하여 하고자 생성 후 30초 후, 3분 이내의 이벤트만을 조회
     *
     * <p> 재발행 후 상태 업데이트
     */
    @Scheduled(fixedRate = 30, timeUnit = TimeUnit.SECONDS)
    public void poll() {
        List<EventOutbox> eventOutboxes = eventOutboxService.readAllPollingEventOutbox();
        
        // 조회한 이벤트 아웃박스들을 발행
        List<CompletableFuture<EventPublishResult>> results = eventOutboxes.stream()
                .map(eventOutbox -> CompletableFuture.supplyAsync(
                                eventPublishTask(eventOutbox),
                                messagePublishTaskExecutor
                        )
                        .thenCompose(future -> future)
                        .exceptionally(ex -> {
                            log.error("Failed to published event \"{}\"", eventOutbox.getEventId(), ex);
                            return EventPublishResult.fail(eventOutbox.getEventId(), ex.getMessage());
                        })
                )
                .toList();
        
        // 모든 이벤트 아웃박스에 대한 메시지 발행이 될 때까지 대기
        CompletableFuture.allOf(results.toArray(new CompletableFuture[0])).join();
        
        // 메시지 발행이 모두 완료되고 난 다음, 이벤트 아웃박스들의 상태를 업데이트
        updateStatus(results);
    }
    
    /**
     * 이벤트 아웃박스 발행 작업을 나타내는 Supplier 메서드
     *
     * @param eventOutbox 이벤트 아웃박스
     * @return 이벤트 메시지 발행 결과 CompletableFuture 객체를 감싸는 Supplier
     */
    private Supplier<CompletableFuture<EventPublishResult>> eventPublishTask(EventOutbox eventOutbox) {
        return () -> eventExternalPublisher.publishRaw(
                eventOutbox.getEventId(), eventOutbox.getEventType(), eventOutbox.getPayload()
        );
    }
    
    /**
     * 이벤트 메시지 발행 결과를 바탕으로 이벤트 아웃박스 상태{@code status}를 업데이트하는 메서드
     *
     * <p> 메서드 호출부에서 모든 이벤트 아웃박스에 대한 메시지 발행이 되기까지 대기하였으므로 결과 객체{@link EventPublishResult}를 즉시 반환
     *
     * <p> 대기 및 실패 이벤트 수를 고려하여 일괄 업데이트 수행
     *
     * @param results 이벤트 메시지 발행 결과 CompletableFuture 리스트
     */
    private void updateStatus(List<CompletableFuture<EventPublishResult>> results) {
        List<String> publishedEventIds = new ArrayList<>();
        List<String> failedEventIds = new ArrayList<>();
        
        results.stream()
                .map(result -> result.join())
                .forEach(result -> {
                    if (result.isSuccess()) {
                        publishedEventIds.add(result.eventId());
                    } else {
                        failedEventIds.add(result.eventId());
                    }
                });
        
        if (!publishedEventIds.isEmpty()) {
            eventOutboxService.updateToPublishedByEventIds(publishedEventIds);
        }
        if (!failedEventIds.isEmpty()) {
            eventOutboxService.updateToFailedByEventsIds(failedEventIds);
        }
    }
}

Summary

  이번 포스팅에서는 트랜잭션 아웃박스 패턴 구현 중 이벤트 폴러(Message Relay)를 적용한 내용을 정리하였다.

  트랜잭션 아웃박스 패턴은 실행 보장을 위해 도입된 패턴으로, 이벤트 아웃박스 폴러를 통해 저장된 아웃박스를 주기적으로 조회하여 메시지 브로커로 발행함으로써 발행 보장을 책임지는 핵심 메커니즘이다.

  RabbitMQ의 Jackson2JsonMessageConverter를 통해 JSON 문자열 형태로 저장된 이벤트를 즉시 발행하고, Publish Confirm 결과를 통해 발행된 이벤트 아웃박스의 상태를 갱신하였다. 이를 통해 외부 메시지 브로커의 기능을 활용한 효율적인 이벤트 발행 구조를 구현할 수 있었다.

  아웃박스 폴링 자체는 비교적 단순했지만, 이 과정에서 Publish Confirm 결과 수신을 위한 CorreltationData 사용 및 메시지 발행을 위한 별도의 쓰레드풀 도입으로 비동기 처리에 대한 고려가 필요했다. 이 과정에서 CompletableFuture를 활용한 비동기 처리 방식에 대해 깊이있게 학습할 수 있었다.

  또한, 아웃박스 조회 및 발행이라는 기본 개념은 단순했지만, 이벤트 유효 시간 관리와 비동기 처리 전략 등 도메인 특성에 따른 설계 고민이 필요했다.

  이번 포스팅을 통해 이벤트 아웃박스에 expired_at 컬럼을 추가하고, 폴링 시 유효 시간을 기준으로 필터링하는 개선 방안을 도출할 수 있었으며, 이를 통해 보다 효율적인 이벤트 발행 구조로 발전 시킬 수 있었다

  이전 포스팅(Transactional Outbox Pattern 도입기 2 - Event Outbox의 저장)에서 비밀번호 초기화/인증번호 이메일 전송 보장을 위한 트랜잭션 아웃박스 패턴 도입 과정 중 도메인 이벤트 발생 후, 이를 아웃박스의 형태로 바꾸어 데이터베이스에 저장하는 과정에 대해 알아보았다.

  도메인 이벤트를 아웃박스로 바꾸어 저장하는 이유는 이후 폴링 작업을 통해 주기적으로 아웃박스를 조회하여 이벤트 발행을 보장하기 위함이다.

  필자는 이벤트 발행을 2단계로 나누어서 처리하였다.

1. 트랜잭션 커밋 이후 즉시 발행 (빠른 이벤트 발행 및 적절한 폴링 주기 사용 목적)
2. 이벤트 아웃박스 폴링을 통한 재발행 (발행 보장)

  Chris Richardson의 Microservice Architecture - Pattern: Transactional outbox에 따르면, 저장소에 저장된 아웃박스를 주기적인 폴링을 통해서 외부 메시지 브로커로 발행한다고 한다. 그러나 필자는 폴링 뿐만 아니라 Spring Event를 통한 트랜잭션 커밋 이후 시점에서 즉시 이벤트를 발행하는 로직을 추가하였다. 이는 이벤트의 빠른 발행과 이벤트를 즉시 발행함으로써 폴링 주기를 늘려 서버 부하를 해결하고자하는 것에 목적이 있다.

  이번 포스팅에서는 이벤트 생성 즉시(트랜잭션 커밋 완료 시점) 이벤트를 발행하는 과정을 도입하게 되며 겪은 경험에 대해 작성해보고자 한다.

기존 이메일 발송 로직

  기존 비밀번호 초기화 요청 및 이메일 인증 기능은 Spring Event를 활용하여 핵심 비즈니스 로직과 이메일 발송 로직을 분리한 구조로 구현되어 있다.

  다만, 이는 동일한 애플리케이션 내부에서 동작하는 이벤트 기반 구조로 논리적 분리만 이루어졌을 뿐 물리적인 분리까지 확장되기는 어려운 상태이다.

  기존 이메일 발송 로직은 다음과 같은 흐름으로 구성되어 있다.

• 핵심 비즈니스 로직 수행 후 Spring Event를 통한 도메인 이벤트 내부 발행
• 이메일 발송용 리스너에서 도메인 이벤트 수신 후 이메일을 발송
• 이메일 발송 시 발생하는 Network I/O 지연을 고려하여 ThreadPoolTaskExecutor 기반 비동기 처리

  위와 같은 구성을 통해 핵심 로직과 부가 로직을 분리하고, 이메일 발송을 비동기적으로 처리하도록 구현하였다.

1) 핵심 비즈니스 로직 수행 후 Spring Event를 통한 도메인 이벤트 발행

  해당 예시는 비밀번호 초기화 요청 로직이다.

@Service
@RequiredArgsConstructor
public class PasswordManagementService {
    private final UserService userService;
    private final PasswordCodeService passwordCodeService;
    private final PasswordEncoder passwordEncoder;
    private final DomainEventPublisher eventPublisher;
    
    /**
     * 비밀번호 초기화 요청 메서드
     *
     * <p> 사용자 확인을 위한 확인용 코드 생성 및 저장
     *
     * <p> 해당 사용자에게 비밀번호 초기화 링크 메일 전송
     *
     * <p> 최종적으로 비밀번호 초기화 요청 이벤트 발행
     *
     * @param email     사용자 이메일
     * @param username  사용자 아이디
     * @throws CustomException AuthErrorType.MISMATCHED_EMAIL_OR_USERNAME - 이메일 또는 아이디가 불일치하는 경우
     */
    @Transactional
    public void requestPasswordReset(String email, String username) {
        if (!userService.existsIncludeDeletedByEmailAndUsername(email, username)) {
            throw new CustomException(AuthErrorType.MISMATCHED_EMAIL_OR_USERNAME);
        }

        if (passwordCodeService.existsCode(email)) {
            throw new CustomException(AuthErrorType.PASSWORD_RESET_TIME_LIMIT);
        }
        
        String code = PasswordCodeGenerator.generate();
        passwordCodeService.save(email, code);
        
        // 비밀번호 초기화 요청 도메인 이벤트 내부 발행
        eventPublisher.publish(new PasswordResetEvent(email, code));
    }
}

  비밀번호 초기화 요청 유즈케이스를 수행하는 PasswordManagementService.requestPasswordReset(...)에서는 사용자의 이메일과 비밀번호 초기화용 인증 코드를 생성 및 저장한 뒤, 해당 정보를 담은 PasswordResetEvent를 발행한다.

  이때 이벤트 발행은 DomainEventPublisher를 통해 수행되며, 내부적으로는 Spring의 ApplicationEventPublisher를 사용하여 애플리케이션 내부 이벤트로 전달된다.

  이후 발행된 이벤트는 @TransactionalEventListener를 통해 이벤트를 구독하는 리스너에서 수신하게 된다.

  비밀번호 초기화 요청 이벤트의 경우, ‘비밀번호 초기화 이메일 발송용 리스너’에서 이벤트를 수신한 뒤 PasswordResetEvent에 포함된 정보를 기반으로 이메일 발송을 수행한다.

2) 이메일 발송용 리스너에서 도메인 이벤트 수신 후 이메일을 발송

@Component
@RequiredArgsConstructor
public class PasswordResetEventEmailListener extends DomainEventListener<PasswordResetEvent> {
    private final PasswordResetEmailSender passwordResetEmailSender;
    
    @Override
    @Async(value = EMAIL_TASK_EXECUTOR)
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleEvent(PasswordResetEvent event) {
        passwordResetEmailSender.send(event.getEmail(), event.getCode());
        // TODO: Transactional Outbox Pattern 적용 시, 이벤트 상태 변경 로직 추가 필요
    }
}

  PasswordResetEventEmailListener는 비밀번호 초기화 시 사용자 인증에 사용되는 확인 코드를 이메일로 발송하는 리스너이다.

  해당 리스너는 PasswordResetEvent를 수신한 뒤, 이벤트에 포함된 정보를 기반으로 PasswordResetEmailSender.send(...)를 호출하여 이메일을 발송하는 단순한 구조로 구성이 되어있다.

  이때, 이메일 발송은 핵심 비즈니스 로직에 대한 부가 로직이다.

해석에 따라 핵심 비즈니스 로직에 포함할 수 있긴 하지만, 이메일 발송은 핵심 비즈니스 로직에 직접적인 영향을 주지 않는 부가 로직으로 보았다.
또한, 이메일 발송은 외부 SMTP 서버와의 통신이 필요한 Network I/O 작업으로, 상대적으로 긴 지연 시간이 발생하기 때문에 핵심 비즈니스 로직과 분리하여 처리하였다.

  또한, 트랜잭션이 롤백될 경우 이메일이 발송되면 안 되기 때문에 트랜잭션 커밋 이후 시점인 TransactionPhase.AFTER_COMMIT 시점에 이벤트를 처리하도록 구성하였다.

  그러나, 이러한 구조는 다음과 같은 한계를 가진다.

• 이벤트 발송 실패 시 자동 재시도 로직이 존재하지 않는다.
• 이벤트 처리 결과를 별도로 저장하지 않기 때문에 상태 추적 또는 재시도가 어렵다.
• 이미 처리된 이벤트는 재사용할 수 없기 때문에 재처리가 사실상 불가능하다.

  즉, 이벤트 처리 실패 시 이메일 발송이 유실될 수 있으며, 이를 보완하기 위해 트랜잭션 아웃박스 패턴을 도입한 것이기도 하다.

3) 이메일 발송 I/O를 고려한 ThreadPoolTaskExecutor 설정

  이메일 발송은 SMTP(Simple Mail Transfer Protocol)을 사용하여 외부 메일 서버와 네트워크 통신을 수행하는 I/O 작업이다. 이 과정에서 TCP 연결 수립, SMTP 핸드셰이크, 수신 서버 MX 레코드 조회(DNS), 메일 서버 내부 처리 과정(큐잉, 스팸 필터링)이 포함되기 때문에 서버 내부에서 행해지는 비즈니스 로직 수행에 비해 많은 지연 시간이 소요된다.

  이러한 특성을 고려하여, 이메일 발송은 Spring Event 기반으로 트랜잭션 커밋 이후 시점으로 분리하고, @Async를 활용해 비동기적으로 처리하도록 구성하였다.

  그러나, @Async만 사용할 경우, 기본 Executor 설정에 따라 스레드 생성이 과도하게 증가하거나 공용 쓰레드풀을 사용할 경우 다른 비동기 작업과 자원을 경쟁하게되어 시스템 부하가 급증할 수 있다.

  따라서, 이메일 발송 작업의 Network I/O 특성을 고려하여 별도의 ThreadPoolTaskExecutor를 도입하였다.

  Brian Goetz’의 저서 Java Concurrency in Practice - 8.2 Sizing Thread Pools에 따르면, I/O 또는 기타 블로킹 작업이 포함된 작업이라면 모든 쓰레드가 항상 스케줄링 가능한 상태로 존재하는 것은 아니기 때문에 작업의 대기시간과 계산 시간의 비율을 측정하여 쓰레드풀 크기를 정해야한다고 명시되어있다.

  즉, 쓰레드는 I/O 작업을 기다리는 동안 CPU는 유휴(idle) 상태가 되므로, 더 많은 쓰레드를 생성하여 전체 처리량을 높이는 것이 효율적이다.

• N_cpu: CPU의 수
• U_cpu: 대상 작업의 목표 CPU 사용률
• W/C: 대기 시간 / 연산 시간

  따라서, 위 공식을 참고해 적정 쓰레드풀 크기를 산정하였다.

  이메일 발송은 I/O Bound 위주 작업이므로 CPU 사용 비중이 낮다. 따라서, CPU를 최대한 활용하여 처리량을 높이기 위해 목표 CPU 사용률(U_cpu)은 1로 설정하였다.

  따라서, 적정_쓰레드_수 = CPU_코어_수 * (1 + wait_time / compute_time)이다.

  측정 결과 이메일 발송에 소요되는 Network I/O 등의 대기 시간은 평균 약 2,400ms, 연산 시간은 평균 약 400ms이다.

  따라서, 적정_쓰레드_수 = CPU_코어_수 * (1 + 2,400 / 400) = CPU_코어수 * 7로 산정할 수 있으며, 이를 기반으로 corePoolSize를 설정하였다.

@EnableAsync
@Configuration
public class AsyncConfig {
    public final static String EMAIL_TASK_EXECUTOR = "emailTaskExecutor";
    
    /**
     * 이메일 발송 작업 수행을 위한 ThreadPoolTaskExecutor 빈
     *
     * <p> 이메일 발송 로직 성능 측정 데이터 기반 corePoolSize 할당
     * <p> 산정 공식: cores * (1 + wait_time / service_time)
     *
     * <p> 예기치 못한 프로세스 종료 시에도, 대기 작업 수행을 위한 graceful shutdown 적용
     *
     * @return 비동기 이메일 발송 작업 용 ThreadPoolTaskExecutor
     */
    @Bean(name = EMAIL_TASK_EXECUTOR)
    public Executor emailTaskExecutor() {
        int cores = Runtime.getRuntime().availableProcessors();
        int corePoolSize = cores * 7;
        int maxPoolSize = corePoolSize * 2;
        
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(100);
        
        executor.setThreadNamePrefix("email-exec-");
        
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        
        return executor;
    }
}

  따라서, 이메일 발송용 비동기 작업 수행을 위한 ThreadPoolTaskExecutor 설정 시 평균 쓰레드 풀 사이즈는 호스트_CPU_수 * 7로 산정하였다. 또한, 이메일 발송 요청 트래픽 급증 상황을 대비하여 maxPoolSize는 corePoolSize * 2로 설정하였다.

  또한, queueCapacity를 설정하여 일시적인 요청 증가 시 작업을 큐에 적재하며, 쓰레드 수를 무제한으로 증가시키는 상황을 방지하였다.

  한편, 현재 구조에서는 재시도 매커니즘이 존재하지 않기 때문에 프로세스가 강제 종료될 경우 진행 중인 이메일 발송 작업이 유실될 가능성이 있다.

  이를 완화하기 위해 graceful shutdown 옵션을 적용하여 애플리케이션 종료 시 진행 중인 작업이 일정 60초 내에 마무리될 수 있도록 구성하였다.

@Component
@RequiredArgsConstructor
public class PasswordResetEventEmailListener extends DomainEventListener<PasswordResetEvent> {
    private final PasswordResetEmailSender passwordResetEmailSender;
    
    @Override
    @Async(value = EMAIL_TASK_EXECUTOR)
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleEvent(PasswordResetEvent event) {
        passwordResetEmailSender.send(event.getEmail(), event.getCode());
        // TODO: Transactional Outbox Pattern 적용 시, 이벤트 상태 변경 로직 추가 필요
    }
}

  빈으로 등록한 이메일 발송 전용 쓰레드 풀은 @Async 어노테이션의 value를 통해 지정 가능하다.

구조 개선

  앞서, 기존 구현에서는 이벤트를 외부로 발행하는 발행자(Publisher)와 발행된 이벤트를 처리하는 소비자(Consumer)가 동일한 애플리케이션에서 동작할 수 밖에 없는 구조였다.

  이로 인해 이메일 발송과 같이 Network I/O 시간이 오래 소요되는 부가 로직이 핵심 비즈니스 로직과 동일한 프로세스에서 처리되었고 이는 아래와 같은 문제점을 야기할 수 있었다.

• 이메일 발송을 위한 외부 시스템 장애 시 서비스 전체로 영향이 전파될 수 있다.
• 부가 로직의 확장 및 독립적인 배포가 불가능
• 이벤트 처리 실패 시 재처리나 실행 보장이 불확실

  따라서, 이러한 문제를 해결하기 위해 외부 메시지 브로커를 도입하여 이벤트 발행자(Publisher)와 소비자(Consumer)를 물리적으로도 분리 가능한 구조로 개선하였다. 이를 통하여 비동기 처리 및 장애 격리의 이점도 얻을 수 있다.

  특히, 메시지 브로커를 통해 재시도, Dead Letter Queue(DLQ), TTL 기반 지연 처리 등 이벤트 처리 보장 및 재시도를 위한 다양한 기능을 활용할 수 있었다.

  이번 포스팅에서는 트랜잭션 아웃박스 패턴 구현 과정 중 트랜잭션 커밋 이후 이벤트를 외부로 즉시 발행하는 구조를 중심으로 다룬다.

  트랜잭션 컴시 이후 즉시 발행된 이벤트는 메시지 브로커(RabbitMQ)로 전달되며, RabbitMQ에서 제공하는 Publish Confirm 기능을 통해 발행한 메시지(이벤트)의 도착 여부를 확인한다. 그 결과에 따라 이벤트 아웃박스의 상태를 갱신한다.

1. RabbitMQ 도입

  외부로 발행되는 이벤트는 메시지 브로커를 통해 전달된다.

  대표적인 메시지 브로커로는 Kafka, AWS SNS/SQS, RabbitMQ 등이 존재한다. 본 프로젝트에서는 비교적 설정이 간단하고, 팀원들이 익숙한 RabbitMQ를 외부 메시지 브로커로 선택하였다.

// build.gradle

/* RabbitMQ */
implementation 'org.springframework.boot:spring-boot-starter-amqp'  

  spring-boot-starter-amqp 디펜던시는 RabbitMQ 관련 클래스와 AutoConfiguration을 제공한다.

--- # rabbitMQ
spring:
  rabbitmq:
    host: ${RABBITMQ_HOST}
    port: ${RABBITMQ_PORT}
    username : ${RABBITMQ_USERNAME}
    password: ${RABBITMQ_PASSWORD}
    publisher-confirm-type: correlated

  application.yml에서 RabbitMQ 관련 환경변수를 바인딩한다.

  host, port, username, password는 RabbitMQ Server와 연결을 위한 기본 설정이다.

  publisher-confirm-type은 Publish Confirm 기능을 활성화하기 위해 필요한 설정으로, RabbitMQ 브로커가 메시지를 정상적으로 수신했는지 여부를 발행자(Publisher)에게 ACK/NACK 형태로 전달한다.

  RabbitMQ - Consumer Acknowledgements and Publisher Confirm에서 소비자(Consumer)에서 메시지 수신 여부를 브로커에 응답하는 Consumer Acknowledgements와 브로커에서 발행자(Publisher)에 메시지 수신 여부를 응답하는 Publisher Confirm에 대해 소개하고 있다.

  트랜잭션 아웃박스 패턴에서 Publisher의 책임은 이벤트(메시지)를 메시지 브로커까지 정상적으로 발행하는 것이다. 따라서, RabbitMQ의 Publisher Confirm 기능을 활용하여 메시지의 정상 발행 여부를 확인해 이벤트 아웃박스의 상태를 갱신한다.

  RabbitMQ의 Publisher Confirm 기능을 사용하기 위해서는 spring.rabbitmq.publisher-confirm-type: correlated 프로퍼티 설정이 필요하다. publisher-confirm-type으로는 3가지 값이 존재한다.

• none: Publisher Confirm 기능을 사용하지 않는다. (기본값)
• simple: 동기 방식으로, 메시지 발행 후 Confirm을 대기
• correlated: 비동기 방식으로, 메시지마다 correlation id를 부여해 Confirm을 처리

  simple 방식은 동기 방식으로 처리하기 때문에 처리량이 낮다는 단점이 있다. 반면, correlated 방식은 비동기적으로 confirm을 처리할 수 있어 높은 처리량을 확보할 수 있다.

  본 구현에서는 이벤트 발행 성능을 고려하여 correlated 방식을 선택하였다.

@Configuration
public class RabbitMQConfig {
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
    
    @Bean
    public MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

  메시지를 발행을 위한 RabbitTemplate과 객체 ↔ JSON 변환을 위한 Jackson2JsonMessageConverter를 빈으로 등록한다.

  이벤트는 아웃박스 레코드의 payload 컬럼에 JSON 문자열 형태로 저장된다. 따라서, 향후 이벤트를 폴링하여 발행할 때 별도의 역직렬화 과정없이 즉시 메시지로 변환하여 전송할 수 있도록 Jackson2JsonMessageConverter를 사용하였다.

RabbitMQ Exchange, Queue, Routing Key, TTL Property 설정

  외부 메시지 브로커로 발행하는 Publisher는 도메인 이벤트 종류에 따라 적절한 Exchange와 Routing Key를 통해 이벤트를 발행하는 역할을 수행한다.

  이를 위해 도메인 이벤트별로 사용할 Exchange, Queue, Routing Key, TTL 값을 프로퍼티로 분리하여 관리하도록 구성하였다.

rabbitmq:
  password-reset:
      queue: ${RABBITMQ_PASSWORD_RESET_QUEUE}
      exchange: ${RABBITMQ_PASSWORD_RESET_EXCHANGE}
      routing-key: ${RABBITMQ_PASSWORD_RESET_ROUTING_KEY}
      ttl: ${RABBITMQ_PASSWORD_RESET_TTL}
  email-verification:
      queue: ${RABBITMQ_EMAIL_VERIFICATION_QUEUE}
      exchange: ${RABBITMQ_EMAIL_VERIFICATION_EXCHANGE}
      routing-key: ${RABBITMQ_EMAIL_VERIFICATION_ROUTING_KEY}
      ttl: ${RABBITMQ_EMAIL_VERIFICATION_TTL}

@Getter
@Setter
@Component
@ConfigurationProperties("rabbitmq")
public class RabbitMQProperties {
    private Attribute passwordReset;
    private Attribute emailVerification;

    @Getter
    @RequiredArgsConstructor
    public static class Attribute {
        private final String queue;
        private final String exchange;
        private final String routingKey;
        private final Long ttl;
    }
}

  각 도메인 이벤트별로 사용할 메시지 브로커 설정을 외부 프로퍼티로 분리함으로써, 환경에 따라 우연하게 설정할 수 있도록 하였다. 하드코딩된 값이 아닌 설정 기반으로 관리함으로써 이벤트 발행 정책을 명확하게 분리하였다.

  수동 지연 큐 전략에서는 하나의 이벤트에 대해 Work Queue / Retry Queue / Dead Letter Queue가 동시에 구성된다.

  이때, 각 Queue는 접미사(postfix)를 통해 역할을 구분하며, 공통으로 사용되는 Exchange, Routing Key는 Queue 이름을 기반으로 파생되는 구조를 사용한다.

  ttl은 Retry Queue에서 재시도 간격을 제어하기 위한 설정으로, 일정 시간 동안 소비되지 않는 메시지를 Dead Letter Exchange(DLX)로 이동하기 위한 TTL 값이다. 이는 향후 이벤트 소비 포스팅에서 자세히 다룰 예정이다.

RabbitMQ Exchange, Routing Key, Queue 바인딩

@Configuration
public class PasswordResetRabbitMQConfig {
    private final RabbitMQProperties.Attribute properties;
    
    public PasswordResetRabbitMQConfig(RabbitMQProperties rabbitMQProperties) {
        this.properties = rabbitMQProperties.getPasswordReset();
    }
    
    @Bean
    public DirectExchange passwordResetExchange() {
        return new DirectExchange(properties.getExchange());
    }
    
    @Bean
    public Queue passwordResetEmailWorkQueue() {
        return QueueBuilder.durable(properties.getQueue() + ".email")
                .withArgument("x-dead-letter-exchange", properties.getExchange() + ".email.retry")
                .withArgument("x-dead-letter-routing-key", properties.getRoutingKey() + ".email.retry")
                .build();
    }
    
    @Bean
    public Binding PasswordResetEmailWorkBinding(Queue passwordResetEmailWorkQueue, DirectExchange passwordResetExchange) {
        return BindingBuilder.bind(passwordResetEmailWorkQueue)
                .to(passwordResetExchange)
                .with(properties.getRoutingKey() + ".email");
    }

    // Retry Queue, DeadLetter Queue 바인딩
}

  위 코드는 비밀번호 초기화 요청 이벤트 발행에 사용되는 Work Queue에 관한 설정이다.

  DirectExchange, Queue, Binding을 각각 빈으로 등록하여 애플리케이션 실행 시 RabbitMQ 자동으로 생성되도록 구성하였다. 물론 RabbitMQ Console을 통해서 수동으로 생성할 수도 있다.

  특히, Work Queue에는 DLX 설정을 추가하여 메시지 처리 실패 시 Retry Queue로 전달될 수 있도록 구성하였다.

  각 요소 구성 시 이름에는 이벤트.사용목적.{큐의_용도}라는 컨벤션을 기반으로 생성하였다. 이는 password-reset.email과 같이 PasswordResetEvent를 email 발송을 위해 처리하는 큐를 네이밍을 통해서 추측할 수 있도록 한다. 또한, password-reset.email.retry와 같이 재시도(수동 지연) 목적의 큐를 나타내기도 한다.

  이를 통해 하나의 도메인 이벤트에 대해 처리해야할 부가 로직 용도에 맞게 Work Queue / Retry Queue / DeadLetter Queue를 일관된 네이밍 규칙으로 관리할 수 있었다.

  이와 같은 구조를 통해 메시지 처리 실패 시에도 Retry Queue를 통한 재시도 흐름을 구성할 수 있었으며, 이에 대한 자세한 소비 처리 로직은 향후 포스팅에서 다룰 예정이다.

RabbitMQ Property Resolver

@Component
@RequiredArgsConstructor
public class RabbitMQPropertyResolver {
    private final RabbitMQProperties properties;
    
    public String getPublishExchange(DomainEvent event) {
        if (event instanceof PasswordResetEvent) {
            return properties.getPasswordReset().getExchange();
        }
        
        if (event instanceof EmailVerificationEvent) {
            return properties.getEmailVerification().getExchange();
        }
        
        throw new IllegalArgumentException("Unsupported event: " + event.getClass().getSimpleName());
    }
    
    public String getPublishRoutingKey(DomainEvent event) {
        if (event instanceof PasswordResetEvent) {
            return properties.getPasswordReset().getRoutingKey() + ".requested";
        }
        
        if (event instanceof EmailVerificationEvent) {
            return properties.getEmailVerification().getRoutingKey() + ".requested";
        }
        
        throw new IllegalArgumentException("Unsupported event: " + event.getClass().getSimpleName());
    }
}

  외부 메시지 브로커로 발행되는 이벤트는 현재 비밀번호 초기화(PasswordReset)과 이메일 인증(EmailVerification) 두 가지로 구성되어있다.

  각 도메인 이벤트는 서로 다른 Exchange와 Routing Key를 통해 적절한 Queue로 라우팅되어야 한다.

  이를 위해 이벤트 타입에 따라 발행하는 Exchange와 Routing Key를 결정하는 RabbitMQPropertyResolver를 정의하였다.

  해당 클래스는 도메인 이벤트를 입력으로 받아, 각 이벤트에 대응하는 메시지 브로커 설정(Exchange, Routing Key)을 반환하는 역할을 수행한다.

  이를 통해 이벤트 발행 로직에서 라우팅 정책을 분리하고, 이벤트 발행 시점에서는 이벤트의 종류만 알면 구체적인 라우팅 전략은 Resolver에게 위임하여 결합도를 낮출 수 있었다.

2. 이메일 발송용 리스너 → 이벤트 발행용 리스너로 변경

  기존 로직에서는 트랜잭션 커밋 후(AFTER_COMMIT) 시점에서 이메일을 직접 발송하는 구조로 구성되어 있었다.

  그러나, 메시지 브로커를 도입하면서 도메인 이벤트의 발행자(Publisher)와 소비자(Consumer)를 분리하였고, 이에 따라 발행자는 이벤트를 메시지 브로커까지 안전하게 전달하는 것에 대한 책임을 가지게 되었다.

  즉, 기존처럼 이메일을 직접 발송하는 것이 아니라, 트랜잭션 커밋 이후 시점에서 이벤트를 외부 메시지 브로커로 발행하는 구조로 변경하였다.

  이를 통해 이벤트 발행과 실제 처리(이메일 발송)의 책임을 분리하고, 이벤트 전달 보장을 위한 구조로 개선하였다.

public interface DomainEventExternalPublisher {
    CompletableFuture<EventPublishResult> publish(DomainEvent event);
}

  우선, 외부 이벤트 발행을 담당하는 DomainEventExternalPublisher 인터페이스를 정의하였다.

  특정 메시지 브로커(RabbitMQ)에 대한 의존성을 추상화하여, 향후 다른 메시지 브로커로 변경하더라도 영향 범위를 최소화하기 위함이다.

  또한, 메시지 브로커로부터 Publish Confirm을 비동기적으로 수신받기 때문에 CompletableFuture<EventPublishResult>를 반환하도록 설계하였다.

  이를 통해 이벤트 발행 결과에 따른 후처리(EventOutbox 상태 갱신)를 비동기적으로 처리할 수 있도록 하였다.

public record EventPublishResult(
        String eventId,
        boolean isSuccess,
        String cause
) {
    public static EventPublishResult success(String eventId) {
        return new EventPublishResult(eventId, true, null);
    }
    
    public static EventPublishResult fail(String eventId, String cause) {
        return new EventPublishResult(eventId, false, cause);
    }
}

  EventPublishResult는 이벤트 발행 결과를 표현하는 객체로, 이벤트 식별자, 발행 성공 여부, 실패 원인을 포함한다.

  해당 객체는 Publish Confirm 결과를 기반으로 생성되며, 이후 이벤트 아웃박스의 상태를 갱신하는 데 사용된다.

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMQEventPublisher implements DomainEventExternalPublisher {
    private final RabbitTemplate rabbitTemplate;
    private final RabbitMQPropertyResolver rabbitMQPropertyResolver;
    
    /**
     * RabbitMQ 메시지브로커로 이벤트를 발행하는 메서드
     *
     * <p> {@code CorrelationData}를 통해 메시지브로커로의 발행 여부를 확인
     *
     * <p> 메시지 발행 중 예외 발생 시 실패 상태 객체를 반환, 성공 시 성공 객체를 반환
     *
     *
     * @param event 도메인 이벤트 객체
     * @return      실행 결과 {@link EventPublishResult}를 감싸고 있는 {@code CompletableFuture} 객체
     */
    @Override
    public CompletableFuture<EventPublishResult> publish(DomainEvent event) {
        CorrelationData correlationData = new CorrelationData(event.getEventId());
        
        CompletableFuture<EventPublishResult> result = correlationData.getFuture().thenApply(confirm -> {
            if (confirm.isAck()) {
                return EventPublishResult.success(event.getEventId());
            } else {
                return EventPublishResult.fail(event.getEventId(), confirm.getReason());
            }
        }).exceptionally(ex -> EventPublishResult.fail(event.getEventId(), ex.getMessage()));
        
        try {
            rabbitTemplate.convertAndSend(
                    rabbitMQPropertyResolver.getPublishExchange(event),
                    rabbitMQPropertyResolver.getPublishRoutingKey(event),
                    event,
                    correlationData
            );
        } catch (Exception e) {
            result.complete(EventPublishResult.fail(event.getEventId(), e.getMessage()));
        }
        
        return result;
    }
}

  RabbitMQEventPublisher는 DomainEventExternalPublisher의 구현체로, RabbitMQ 메시지 브로커로 이벤트를 발행하는 역할을 수행한다.

  rabbitTemplate.convertAndSend(...) 메서드를 통해 이벤트를 발행하고, CorrelationData를 함께 전달하여 Publish Confirm 결과를 비동기적으로 수신한다.

  CorrelationData는 메시지 발행 시 고유 식별자인 Correlation Id(= eventId)를 함께 전달하여, 비동기적으로 수신되는 Publish Confirm 결과를 특정 메시지와 매핑하기 위한 객체이다.

  RabbitMQ에 메시지가 브로커에 정상적으로 도착했을 경우 ACK, 실패하 경우 NACK 신호를 반환하며, CorrelationData.getFuture()를 통해 해당 결과를 CompletableFuture 형태로 수신할 수 있다.

  따라서, CompletableFuture<CorrelationData.Confirm>에 .thenApply(...)를 적용하여 Confirm 결과를 EventPublishResult로 변환하는 후처리 로직을 구성하였다.

  Confirm.isAck()이 true인 경우 메시지가 브로커에 정상적으로 전달된 것이므로 EventPublishResult.success(...)를 반환하고, false인 경우 EventPublishResult.fail(...)을 반환하도록 설계하였다.

  또한, Publish Confirm 처리 과정에서 발생하는 예외는 .exceptionally(...)를 통해 처리하며, 메시지 발행 과정(convertAndSend)에서 발생하는 동기 예외는 try-catch 블록을 통해 별도로 처리하였다.

  RabbitTemplate.convertAndSend(...)의 첫 번째, 두 번째 인자로 메시지를 발행할 Exchange와 Routing Key를 명시해야 한다.

  도메인 이벤트별로 서로 다른 Queue로 라우팅 하기 위해 각기 다른 Exchange와 Routing Key가 필요하며, RabbitMQPropertyResolver를 통해 이벤트에 맞는 설정 값을 조회하여 메시지를 발행하도록 구성하였다.

@Component
@RequiredArgsConstructor
public class DomainEventPublishListener implements DomainEventListener {
    private final DomainEventExternalPublisher eventExternalPublisher;
    private final EventOutboxService eventOutboxService;
    private final Executor messagePublishTaskExecutor;
    
    @Override
    @Async(value = MESSAGE_PUBLISH_TASK_EXECUTOR)
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleEvent(DomainEvent event) {
        eventExternalPublisher.publish(event)
                .thenAcceptAsync(result -> {
                    if (result.isSuccess()) {
                        eventOutboxService.updateToDoneByEventId(result.eventId());
                    } else {
                        eventOutboxService.updateToFailedByEventId(result.eventId());
                    }
                }, messagePublishTaskExecutor);
    }
}

@EnableAsync
@Configuration
public class AsyncConfig {
    public final static String MESSAGE_PUBLISH_TASK_EXECUTOR = "messagePublishTaskExecutor";
    
    /**
     * 이벤트 메시지 발행을 위한 ThreadPoolTaskExecutor 빈
     *
     * @return 비동기 이메일 발송 작업 용 ThreadPoolTaskExecutor
     */
    @Bean(name = MESSAGE_PUBLISH_TASK_EXECUTOR)
    public Executor messagePublishTaskExecutor() {
        int cores = Runtime.getRuntime().availableProcessors();
        int corePoolSize = cores * 2;
        int maxPoolSize = cores * 4;
        
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(100);
        
        executor.setThreadNamePrefix("message-publish-exec-");
        
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        
        return executor;
    }
}

@Service
@RequiredArgsConstructor
public class EventOutboxService {
    private final EventOutboxRepository eventOutboxRepository;
    
    // ...

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void updateToDoneByEventId(String eventId) {
        eventOutboxRepository.updateStatusPublishedByEventId(eventId);
    }
    
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void updateToFailedByEventId(String eventId) {
        eventOutboxRepository.updateStatusFailedByEventId(eventId);
    }
}

  이제 트랜잭션 커밋 완료(AFTER_COMMIT) 시점에서 EventPublishResult 결과에 따라 이벤트 아웃박스의 상태를 갱신해야 한다.

  이를 위해 eventExternalPublisher.publish(event)의 반환값인 CompletableFuture<EventPublishResult>를 처리하기 위해 .thenAcceptAsync(Consumer, Executor)를 적용하여 비동기적으로 후처리를 수행한다.

  EventPublishResult가 성공 객체인 경우 이벤트 아웃박스의 상태는 성공(PUBLISHED), 실패 객체일 경우에는 실패(FAILED) 상태로 갱신한다.

  이때, 이벤트 발행 처리 성능을 고려하여 이벤트 메시지 발행 전용 ThreadPoolTaskExecutor를 새로 정의하였다. 이벤트 발행 자체에는 많은 시간이 소요되지 않기 때문에 corePoolSize는 사용 가능 코어 수의 2배로 설정하고, 트래픽이 집중적이로 몰릴 경우를 대비해 최대 4배로 늘어나도록 설정하였다.

이메일 발송 후, 트랜잭션 아웃박스 상태가 갱신되지 않던 이슈

  그러나, 이 과정에서 트랜잭션이 적용되지 않는 문제가 발생하였다.

  DomainEventPublishListener.handleEvent(DomainEvent)는 @Async를 통해 별도의 쓰레드풀에서 동작하며, 이후 .thenAcceptAsync(...) 및 Publish Confirm 처리 과정에서도 추가적인 쓰레드 전환이 발생한다.

  Spring의 트랜잭션은 ThreadLocal 기반으로 동작하기 떄문에 이와 같이 쓰레드가 변경될 경우 기존 트랜잭션 컨텍스트가 전파되지 않는다.

  즉, 동일한 메서드에 @Transactional을 추가하더라도 비동기 실행 구간에서는 트랜잭션이 적용되지 않는다.

  따라서, 이벤트 아웃박스 상태 갱신 로직은 별도의 트랜잭션에서 독립적으로 수행할 수 있도록 구성하였다.

  EventOutboxService 도메인 서비스 클래스의 상태 갱신 메서드에 @Transactional(propagation = Propagation.REQUIRES_NEW)를 적용하여, 각 상태 변경 작업이 새로운 트랜잭션에서 실행되도록 하였다.

  이를 통해 비동기 처리 환경에서도 이벤트 아웃박스 상태를 안정적으로 갱신할 수 있도록 개선하였다.

결론

  이번 작업은 트랜잭션 아웃박스 패턴을 적용하여 Chris Richardson이 소개한 이벤트 폴링 작업 외에도 트랜잭션 커밋 이후(AFTER_COMMIT) 시점에 이벤트를 외부 메시지 브로커로 즉시 발행하는 구조를 추가로 도입하였다.

  이는 Spring Event가 제공하는 트랜잭션 시점 기반 이벤트 처리 장점을 활용하면서, 이벤트를 즉시 발행하여 폴링 대상 이벤트 수를 줄이고 전체 처리 효율을 개선하기 위한 설계이다.

  기존 구조에서는 Spring Event를 활용하여 핵심 비즈니스 로직과 부가 로직을 분리하였지만, 동일한 애플리케이션에서 동작할 수 밖에 없는 노리적인 분리이었기에 물리적 분리로 확장 가능성은 존재하지 않았다.

  따라서, 이메일 발송과 같이 Network I/O 기반의 부가 로직이 핵심 비즈니스 로직과 동일한 프로세스에서 처리되어 성능 부담과 장애 전파의 가능성을 내포하고 있었다.

  이번 구조 개선을 통해 이벤트 발행자(Publisher)와 소비자(Consumer)를 분리하고, Publisher는 이벤트를 메시지 브로커까지 안전하게 전달하는 역할에 집중하도록 구성하였다.

  현재 서비스는 모놀리식 구조이지만, 이번 설계를 통해 향후 부가 로직을 별도의 서비스로 분리하거나 확장하는 것이 용이한 구조를 갖추게 되었다.

  다음 포스팅에서는 트랜잭션 아웃박스 패턴의 핵심인 이벤트 폴링 기반 발행 보장 메커니즘에 대해 다룰 예정이다.

  트랜잭션 아웃박스 패턴은 아래와 같은 처리 흐름을 가진다.

• 도메인 이벤트 발행 (내부 발행)
• 도메인 이벤트 리스너 - 이벤트 저장 (아웃박스 변환)
• 도메인 이벤트 리스너 - 이벤트 발행 (발행 후 아웃박스 상태 변경)
• 이벤트 폴러 - 아웃박스 조회 후 이벤트 발행
• 메시지 브로커
• 이벤트 소비자

  트랜잭션 아웃박스 패턴의 구현은 변형되어 다양한 방식이 존재한다. Chris Richardson의 Microarchitecture - Pattern: Transactional outbox에 따르면, 이벤트 발행은 메시지 릴레이(Message Relay = Event Poller)가 담당한다.

  그러나, 필자는 Spring Event의 @TransactionalEventListener를 통해서 트랜잭션 시점에 따른 이벤트 기록, 이벤트 외부 발행을 진행하여 이벤트 폴러 외에도 이벤트 외부 발행 기능을 담당하는 주체가 하나 더 존재한다. 이러한 구조를 선택한 이유는 향후 포스팅에서 서술한다.

  이번 포스팅에서 집중할 부분은 ‘이벤트 저장’이다.

  Spring Event를 통해서 내부 발행된 이벤트는 이벤트 저장용 리스너에 의해 Event Outbox로 변환되어 저장된다.

  본론에 들어가기에 앞서 ‘이벤트 저장’ 로직의 흐름은 다음과 같다.

• DomainEvent 상위 추상 클래스 정의 및 Spring Event를 사용한 내부 발행
• 각 비즈니스 로직 내 이벤트 발행 로직 적용 및 도메인 이벤트 구현
• @TransactionalEventListener를 사용한 이벤트 기록용 리스너 정의
• 이벤트 기록용 리스너에서 이벤트를 아웃박스 엔티티(EventOutbox) 형태로 전환하여 저장

  크게 위와 같은 흐름으로 진행되며 이를 적용하기 위해 여러 클래스들을 구현하였다.

  본론에서 Event 클래스, Event Outbox 엔티티의 구조와 저장 흐름에 대해 설명하고자 한다.

DomainEvent 클래스와 Spring Event를 사용한 내부 이벤트 발행

  Spring Modulith - Working with Application Events에서는 ApplicationEventPublisher을 통해서 이벤트를 발행하여 클래스 간 결합도를 낮출 수 있다고 명시되어 있다.

  또한, ApplicationEventPublisher를 통해서 발행한 이벤트는 @EventListener 또는 @ApplicationModuleListener, @TransactionalEventListener를 통해서 이벤트 리스닝이 가능하다.

  특히, @TransactionalEventListener는 트랜잭션 단계(Phase)에 따라 호출되는 이벤트 리스너로, 트랜잭션 단계는 아래와 같이 4가지 종류가 존재한다.

• AFTER_COMMIT: 트랜잭션 커밋 후 (default)
• AFTER_COMPLETION: 트랜잭션 종류 후 (커밋/롤백에 상관없이)
• AFTER_ROLLBACK: 트랜잭션 롤백 후
• BEFORE_COMMIT: 트랜잭션 커밋 전

  이벤트의 저장 단계에서 필요한 트랜잭션 단계(시점)는 커밋 전(BEFORE_COMMIT)이다.

  트랜잭션이 커밋되기 전 이벤트(아웃박스)가 저장소에 기록되어야지만 향후 저장된 이벤트(아웃박스)를 조회하여 메시지 발행이 가능하다.

DomainEvent

  DomainEvent는 특정 비즈니스 도메인에서 이벤트가 발생하였을 때 발행할 모든 이벤트 상위 추상 클래스이다.

  ApplicationEventPublisher에서 발행한 이벤트는 리스너에서 인자로 명시한 클래스 타입에 따라 이벤트를 가져와 처리하게 된다. DomainEvent라는 상위 추상 클래스를 정의하여 모든 이벤트에 일관된 내부 발행 로직을 적용하고, 이벤트 리스너에서는 공통 로직은 DomainEvent, 개별 실행 로직은 구체적 타입을 명시하여 처리한다.

  또한, 로그 기록과 같은 추적 작업에 DomainEvent에 대한 리스너를 추가하여 사용하는 등의 작업도 가능하다.

@Getter
public abstract class DomainEvent {
    private final ZonedDateTime createdAt;
    private final String eventId;
    
    public DomainEvent() {
        this.createdAt = ZonedDateTime.now(ZoneId.of("Asia/Seoul"));
        this.eventId = UlidCreator.getUlid().toString();
    }
}

  DomainEvent는 모든 클래스의 상위 추상 클래스로, 모든 클래스가 포함하고 있어야할 속성을 가진다.

  이벤트 발행 시각을 나타내는 createdAt과 개별 이벤트 고유번호를 나타내는 eventId를 가진다.

  이벤트마다 해당 이벤트가 유효한 시간이 다르며, 이벤트를 추적하기 위해서는 이벤트를 식별하기 위한 식별자가 필요하기 때문에 위와 같은 속성을 정의하였다.

  createdAt 속성은 서버 환경에 따라 변하는 것을 방지하기 위하여 ZonedDateTime 타입을 사용하여 코드 내에 TimeZone(Asia/Seoul)을 명시하였다.

  eventId는 이벤트를 식별하기 위한 고유번호로 충돌 방지와 효율적인 저장 방식을 고려하여 ULID를 사용하였다.

  이벤트 식별자로 UUID대신 ULID를 사용한 이유는 이전 포스팅에서 정리하였듯 이벤트 아웃박스의 저장에 있어 성능적인 부분을 고려하였기 때문이다.

  고유 식별자의 크기를 줄여 이벤트 아웃박스 테이블이 차지하는 페이지의 크기를 줄이고, Key가 시간에 따라 순차적으로 증가하는 양상을 보여 레코드 삽입 시 인덱스 트리 구조 갱신을 최소화하기 위해 ULID를 사용하였다.

  이벤트 아웃박스는 모든 발생 이벤트들이 메시지 브로커로 발행되기 전 임시로 저장되기 때문에 다양한 도메인에서 많은 이벤트들이 발생하게 되어 이러한 성능적 개선점을 도입하게 되었다.

PasswordResetEvent

@Getter
public class PasswordResetEvent extends DomainEvent {
    private final String email;
    private final String code;
    private final ZonedDateTime expiredAt;
    
    public PasswordResetEvent(String email, String code, int expiration) {
        super();
        this.email = email;
        this.code = code;
        this.expiredAt = getCreatedAt().plusSeconds(expiration);
    }
}

  PasswordResetEvent는 비밀번호 초기화 요청을 나타내는 이벤트 클래스로 DomainEvent를 구현한 클래스이다. 트랜잭션 아웃박스 패턴을 적용하는 모든 이벤트 클래스들이 DomainEvent를 상속받아 구현한다.

  DomainEvent에 속하는 공통 적용 속성을 제외하고, 비밀번호 초기화 이벤트에 맞는 속성 email, code(인증코드), expiredAt(만료 기한)을 가진다.

cf. 이벤트의 구조(속성)

  이메일과 같은 부가로직 수행이 아닌, CQRS 환경이나 여러 모듈 간 데이터를 공유하여 저장하고 있는 MSA 환경일 경우 데이터베이스 갱신 작업 전파를 위한 트랜잭션 아웃박스 패턴을 구현하기도 한다.

  실제로 MSA로 운영되는 서비스들의 기술블로그에서는 이러한 데이터 갱신 이벤트 전파의 목적으로 트랜잭션 아웃박스 패턴을 적용하는 사례도 보았다.

  해당 경우에는 부가 로직 실행이 아닌 ‘갱신된 데이터의 변경사항을 확인’하는 것이 목적이기에 제로 페이로드(Zero Payload) 방식을 사용하기도 한다.

  제로 페이로드 방식은 데이터가 갱신된 A 모듈에서는 엔티티의 주키(PK)만 담은 이벤트를 발행한다. 이후, 이벤트를 수신한 B 모듈에서는 갱신된 엔티티의 PK를 확인하고 해당 키를 통해 A 모듈에 엔티티 조회 요청을 보낸다. B 모듈은 갱신된 엔티티를 응답받아 데이터를 갱신하는 등의 작업을 수행한다.

  단순히, 주키만 전달하는 것이 아닌 갱신된 엔티티의 필드나 이유 등을 나누고 이를 문서화하여 별도로 관리하여 필요에 맞게 유연하게 변경하여 사용하기도 한다.

비밀번호 초기화 요청 로직

@Service
@RequiredArgsConstructor
public class PasswordManagementService {
    private final UserService userService;
    private final PasswordCodeService passwordCodeService;
    private final DomainEventPublisher eventPublisher;
    
    @Value("${auth.expiration.password-reset}")
    private int passwordResetExpiration;
    
    /**
     * 비밀번호 초기화 요청 메서드
     *
     * <p> 사용자 확인을 위한 확인용 코드 생성 및 저장
     *
     * <p> 해당 사용자에게 비밀번호 초기화 링크 메일 전송
     *
     * <p> 최종적으로 비밀번호 초기화 요청 이벤트 발행
     *
     * @param email     사용자 이메일
     * @param username  사용자 아이디
     * @throws CustomException AuthErrorType.MISMATCHED_EMAIL_OR_USERNAME - 이메일 또는 아이디가 불일치하는 경우
     */
    @Transactional
    public void requestPasswordReset(String email, String username) {
        if (!userService.existsIncludeDeletedByEmailAndUsername(email, username)) {
            throw new CustomException(AuthErrorType.MISMATCHED_EMAIL_OR_USERNAME);
        }

        if (passwordCodeService.existsCode(email)) {
            throw new CustomException(AuthErrorType.PASSWORD_RESET_TIME_LIMIT);
        }
        
        String code = PasswordCodeGenerator.generate();
        passwordCodeService.save(email, code, passwordResetExpiration);
        
        // PasswordResetEvent를 생성해 발행
        eventPublisher.publish(new PasswordResetEvent(email, code, passwordResetExpiration));
    }
    // ...
}

  비밀번호 초기화 요청 로직에서는 비밀번호 초기화 이메일 전송 시, 비밀번호 초기화 가능한 링크와 인증번호를 전송하게 된다.

  인증번호 생성이 완료된 후, 이메일을 보내기 위한 PasswordResetEvent를 생성해 발행하게 된다.

  이벤트 기반 구조를 도입하여 메서드 응집도 및 비밀번호 초기화 요청 유즈케이스와 이메일 발행 로직간 강결합도를 줄일 수 있다.

DomainEventPublisher

@Component
@RequiredArgsConstructor
public class DomainEventPublisher {
    private final ApplicationEventPublisher applicationEventPublisher;
    
    public void publish(DomainEvent event) {
        applicationEventPublisher.publishEvent(event);
    }
}

  DomainEventPublisher는 ApplicationEventPublisher를 감싼 클래스로 내부 이벤트 발행을 담당한다.

  내부 이벤트 발행은 Spring Event의 Event 발행 구조를 의미한다.

EventOutbox

  이벤트 아웃박스는 DomainEvent를 데이터베이스에 저장하기 위한 아웃박스로 변환한 엔티티를 의미한다.

@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Entity(name = "event_outbox")
public class EventOutbox {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "event_id", nullable = false, length = 26)
    private String eventId;
    
    @Column(name = "event_type", nullable = false)
    private String eventType;
    
    @Column(nullable = false)
    private String payload;
    
    @Column(nullable = false)
    @Enumerated(EnumType.STRING)
    private EventOutboxStatus status;
    
    @Column(name = "created_at", nullable = false)
    private LocalDateTime createdAt;
    
    @Column(name = "last_retried_at")
    private LocalDateTime lastRetriedAt;
    
    @Column(name = "fail_count")
    private int failCount;
    
    @Builder
    public EventOutbox(String eventId, String eventType, String payload, LocalDateTime createdAt) {
        this.eventId = eventId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = createdAt;
        this.status = EventOutboxStatus.WAITING;
    }
    
    public static EventOutbox fromEvent(DomainEvent event, String payload) {
        return EventOutbox.builder()
                .eventId(event.getEventId())
                .eventType(event.getClass().getName())
                .payload(payload)
                .createdAt(event.getCreatedAt().toLocalDateTime())
                .build();
    }
}

  이벤트 아웃박스의 각 필드에 대한 설명은 다음과 같다.

1. Long id

  이벤트 아웃박스의 고유번호로 Long(BIGINT) 타입으로 저장된다.

  주 키(Primary Key)의 경우 키-레코드 쌍으로 저장이되기 때문에 주키는 데이터베이스에서 생성하는 순차성을 완전히 보장하는 전략으로 선택하였다.

2. String eventId

  이벤트를 식별하기 위한 고유번호로 ULID를 문자열 형태로 변환하여 저장한다.

  Primary Key는 데이터베이스에서 키를 생성하는 IDENTITY 타입을 사용했기 때문에 애플리케이션 단에서 키를 알기 어렵다.

  따라서, 애플리케이션 단계에서 별도의 키를 생성하여 이벤트를 추척할 수 있도록 하였다.

  또한, eventId를 통한 빠른 조회가 가능하도록 eventId 컬럼을 통한 보조 인덱스에도 사용된다.

event_id 컬럼에 대한 unique 설정을 통해 자동으로 인덱스를 생성해 사용할 수도 있다.

3. String eventType

  이벤트의 클래스 타입을 나타내는 속성이다.

  이벤트는 아웃박스 형태로 저장될 때, 이벤트 자체는 JSON 형태의 문자열로 직렬화되어 저장된다.

  이후, 이벤트 아웃박스를 조회할 때 이벤트를 다시 복구하기 위해서 타입이 필요하다. 따라서, 이벤트의 클래스 타입을 저장한다.

적용된 외부 메시지브로커(RabbitMQ)는 JSON 형태의 문자열 메시지도 발행이 가능하기 때문에 실제로는 이벤트를 조회할 때 별도로 변환하는 과정은 존재하지 않으나, 이벤트 추적 및 향후 확장을 위하여 이벤트 클래스 타입을 동시에 저장한다.

4. String payload

  실제로 도메인 이벤트 객체가 JSON 형태의 문자열로 직렬화되어 저장되는 컬럼이다.

  해당 컬럼이 이벤트를 나타내는 핵심 페이로드이며, 이를 역직렬화하거나 문자열 그대로 발행하는 등의 작업에 사용된다.

5. EventOutboxStatus status

public enum EventOutboxStatus {
    WAITING, PUBLISHED, FAILED
}

  이벤트 아웃박스의 상태를 나타내는 클래스이다.

  이벤트 발행 대기, 발행 성공, 발행 실패의 상태를 가진다.

  필요에 따라 상태를 세분화할 수 있다.

6. LocalDateTime createdAt

  이벤트의 생성 시간을 나타내는 컬럼이다.

  DomainEvent 추상 클래스의 createdAt과 동일하다.

7. LocalDateTime lastRetriedAt

  마지막으로 재시도한 시각을 나타내는 컬럼이다.

  특정 이벤트의 경우에는 이벤트 자체의 유효기간이 존재하기도 하며, 향후 이벤트 추적을 위해서도 사용되는 컬럼이다.

8. int failCount

  이벤트의 외부 발행 시도 실패 횟수를 나타내는 컬럼이다.

  각 이벤트의 성격에 맞게 실패 횟수를 통한 발행 제어가 가능하다. 정확하게는 이벤트 폴러에서 특정 횟수 이상 실패한 이벤트는 조회하지 않도록하여 조회 부담을 줄인다.

DomainEventRecordListener

public interface DomainEventListener {
    void handleEvent(DomainEvent event);
}

@Component
@RequiredArgsConstructor
public class DomainEventRecordListener implements DomainEventListener {
    private final EventOutboxService eventOutboxService;
    private final ObjectMapper objectMapper;
    
    @Override
    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
    public void handleEvent(DomainEvent event) {
        try {
            String payload = objectMapper.writeValueAsString(event);
            
            eventOutboxService.save(EventOutbox.fromEvent(event, payload));
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to parse Event - " + event.getEventId(), e);
        }
    }
}

  DomainEventRecordListener는 이벤트를 아웃박스로 변환하여 event_outbox 테이블에 기록하기 위한 이벤트 리스너이다.

  @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)를 사용하여 트랜잭션이 커밋되기 전에 이벤트가 저장되도록 한다.

  부가로직이라 할지라도 이벤트가 아웃박스 형태로 저장이 되어야지만 이후 다른 모듈로 전달되어 부가 로직들을 실행할 수 있다. 따라서, 트랜잭션 커밋 이전 시점에 실행하여 핵심 비즈니스 로직과 이벤트 아웃박스 저장 로직 실행의 원자성을 보장한다.

  각 이벤트는 ObjectMapper를 사용해 JSON 형태의 문자열로 변환되어 EventOutbox.payload 속성으로 저장된다.

이벤트 저장 흐름 정리

  위 과정을 통하여 트랜잭션 아웃박스 패턴 중 ‘이벤트 저장’ 로직에 관여하는 주요 클래스들에 대해 알아보았다.

  저장 흐름에서 핵심은 Spring Event를 사용하여 트랜잭션이 커밋되기 전에 이벤트를 아웃박스 형태로 저장하는 것이다.

  내부 이벤트를 발행하는 과정에서 DomainEvent 상위 추상 클래스를 적용하여 각 이벤트마다 별도의 기록용 리스너를 사용하는 것이 아니라 부모 타입인 DomainEvent를 리스닝하는 이벤트 리스너 하나만 정의하여 기록 로직을 통합하였다.

  각 비즈니스 로직은 부가 로직을 직접 수행하는 것이 아닌 DomainEvent를 상속받은 이벤트 객체를 생성하여 내부로 발행하는 책임을 가지게 된다.

  내부로 발행된 이벤트는 우선 트랜잭션 커밋 전에 EventOutbox 엔티티로 변환되어 저장소에 저장된다. 이때, @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT) 어노테이션을 통해 비즈니스 로직과 이벤트 아웃박스 저장을 원자적으로 실행한다.

마무리하며

  해당 포스팅에서는 트랜잭션 아웃박스 패턴을 구현하는 중 ‘이벤트 저장’ 흐름에 대해 알아보았다.

  다음 포스팅에서는 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)을 사용하여 커밋이 완료된 후 이벤트를 발행하는 로직을 구현한 경험에 대해 서술해보고자 한다.

  여행 기록 관리 플랫폼 ‘여기가’ 프로젝트를 진행하며, 비밀번호 초기화 이메일 전송 기능을 담당하게 되었다.

  기능 구현을 완료한 뒤 이메일 전송 보장 기능을 도입해야하는 추가 이슈가 발생하였다. 여러 방법들을 모색하던 중 트랜잭션 아웃박스 패턴(Transactional Outbox Pattern)을 알게되었다.

  트랜잭션 아웃박스 패턴에 대한 기술 블로그 및 예제 코드를 찾아보면 EDA(Event Driven Architecture)와 함께 사용되는 경우를 많이 보았다.

  학부생 규모의 프로젝트를 많이 진행하면서 모놀리식 단일 모듈 구조를 많이 사용하였다. 따라서, 이벤트를 활용해본 경험이 거의 전무하였기에 “트랜잭션 아웃박스 패턴을 사용할 때, 왜 EDA도 함께 사용하는가?” 라는 의문이 들었다.

  결론부터 말하자면, 트랜잭션 아웃박스 패턴을 사용할 때 EDA를 같이 사용하는 것이 아니라 EDA를 사용할 때 트랜잭션 아웃박스 패턴을 사용하는 것이다.

  앞서 이벤트를 통한 시스템 결합도 낮추기 포스팅에서 이벤트 기반 아키텍처가 왜 필요한지에 대해서 다루었다.

  이벤트는 시스템의 결합도를 낮추고, 장애를 격리시킨다는 장점으로 인하여 MSA 환경에서 많이 사용된다.

  이메일 전송과 같이 외부 API를 사용하거나 입출력에 많은 시간이 소요되는 작업은 이벤트 기반 아키텍처를 일부 도입하여 해당 기능을 수행하는 주체를 분리하는 것이 성능 및 처리 효율성 향상에 큰 도움이 된다.

  그러나, 이벤트를 통한 처리 시 발생하는 문제가 존재한다.

• 메시지 전송 전 유실 문제
• 핵심 로직과 메시지 발송이 같은 트랜잭션으로 묶여야하는 문제

  위 2가지 문제가 EDA에서 Transactional Outbox Pattern을 사용하게 되는 이유이다.

메시지 유실 문제

  메시지 브로커는 내부에서 재시도 로직이나 메시지 영속화 등 다양한 기능을 제공한다. 즉, 메시지 브로커로 메시지(이벤트)가 도착하면 내부에서 제공하는 기능으로 메시지를 기억하거나 재시도 등의 작업을 수행할 수 있다.

  그러나, 서버 애플리케이션에서 메시지 브로커로 메시지를 전송하는 도중에 메시지가 유실되어버린다면 더이상 처리할 수 없게 된다.

  애플리케이션 내부에서 처리되는 이벤트라면 처리 흐름 안에 이벤트가 여전히 존재하기때문에 유실될 가능성이 존재하지 않는다. 그러나, MSA 환경에서는 애플리케이션이 독립적으로 구성되어 있다. 주로 메시지 브로커를 통해 외부로 이벤트를 전달하는 이유는 해당 이벤트를 처리해야하는 부가 로직이 존재하기 때문이다.

  따라서, 각 애플리케이션 간 데이터를 이벤트 기반으로 주고받으며 Kafka, RabbitMQ, AWS SNS/SQS와 같은 외부 메시지 브로커를 이용한다. 외부 메시지 브로커가 일시적으로 장애가 발생하거나 메시지큐(버스)가 가득 차 더이상 메시지를 처리할 수 없을 때 단순히 이벤트를 발송하기만 하면 해당 이벤트는 유실되어 버린다.

핵심 로직과 메시지 발송이 같은 트랜잭션으로 묶여야하는 문제

  비밀번호 초기화 기능의 핵심 로직인 ‘임시 비밀번호 발급’과 부가 로직인 ‘비밀번호 초기화 이메일 전송’은 하나의 트랜잭션으로 묶어 실행되어야지만 정상적으로 동작하는 기능이다.

비밀번호 초기화 이메일 전송은 핵심 로직에 포함되기는 하나, 이메일 전송의 측면에서 부가 로직으로 설명하였다.

  이렇듯 대부분 DB를 업데이트하는 작업은 트랜잭션과 함께 메시지를 발행해야한다.

  만약, DB 업데이트와 메시지 발행을 하나의 트랜잭션으로 묶지 않으면 다음과 같은 문제가 발생할 수 있다.

• 임시 비밀번호 갱신 실패, 비밀번호 초기화 메일 전송 성공
• 임시 비밀번호 갱신 성공, 비밀번호 초기화 메일 전송 실패

  비밀번호 초기화 메일 전송 후 에러가 발생하여 임시 비밀번호 갱신에는 실패하여 임시 비밀번호 갱신 쿼리가 롤백된다면, 사용자들은 초기화되었다는 비밀번호로 로그인을 시도하겠지만 로그인은 실패하게 된다.

  이와 반대로, 임시 비밀번호는 갱신하였는데, 비밀번호 초기화 메일이 전송되지 않았다면 사용자들은 새로 바뀐 비밀번호를 알지 못해 로그인을 하지 못하는 상황이 발생할 것이다.

  이 문제를 해결하는 방법이 트랜잭션 아웃박스 패턴(Transactional Outbox Pattern)이다.

Transactional Outbox Pattern

  이벤트 기반 아키텍처에서 외부 메시지 브로커를 사용하여 다른 애플리케이션으로 메시지를 전달할 경우 ‘메시지 전송 전 유실’, ‘핵심 로직과 메시지 발송이 동일 트랜잭션 내 위치’와 같은 문제가 존재한다는 것을 확인하였다.

  이를 해결하기 위한 방법이 트랜잭션 아웃박스 패턴(Transactional Outbox Pattern)이다.

  트랜잭션 아웃박스 패턴의 핵심은 이벤트를 데이터베이스에 아웃박스(Outbox)로 저장하는 것이다. 또한, 이 아웃박스의 저장을 하나의 트랜잭션으로 핵심 로직의 실행과 하나의 트랜잭션으로 묶는다.

  이를 통하여 핵심 로직이 완전하게 실행된 경우에만 아웃박스가 존재하게 된다. 따라서, 이후 서버에서는 메시지 릴레이(Message Relay)를 통하여 주기적으로 아웃박스를 조회하여 메시지 브로커로 발행한다.

Outbox

  아웃박스(Outbox)는 사전적인 의미로 ‘전송 중이거나 전송에 실패한 메시지가 점시 머무는 보관함’을 의미한다.

  해당 의미 그대로, 외부로 발행해야할 이벤트들이 아웃박스 형태로 변환되어 데이터베이스에 잠시 머무르게 되는 것이다.

이벤트의 저장

  트랜잭션 아웃박스 패턴의 핵심은 이벤트를 저장하는 것이다. 정확하게는 아웃박스의 형태로 저장하는 것이다.

  아웃박스를 저장하기 위한 이벤트 저장소(데이터베이스)를 고민해야한다.

  아웃박스에는 이벤트의 정보와 아웃박스 자체에 대한 메타 데이터 등에 대한 내용만 가지고 있기 때문에 작은 단위로 저장되며, 이벤트는 고속으로 처리되어야 하기 때문에 RDBMS가 아닌 다른 데이터베이스를 사용해야한다고 생각할 수 있다.

  그러나, 우아한 기술 블로그 - 회원 시스템 이벤트기반 아키텍처 구축하기 포스팅에서 같은 RDBMS를 사용해도 괜찮다는 내용을 확인할 수 있었다. 이벤트 저장소와 도메인 저장소를 다른 종류의 데이터베이스로 사용할 경우, 두 저장소에 대한 트랜잭션을 처리해야하나 다종 데이터베이스의 분산 트랜잭션을 구현하는 것은 굉장히 어려운 일이라고 설명하고 있다.

  따라서, 도메인 저장소와 이벤트 저장소를 동일한 DBMS로 사용한다면 트랜잭션 처리는 DBMS에 믿고 맡길 수 있으며, 단일 저장소를 사용하여 쓰기량 및 읽기량에 대한 성느적 리스크는 스케일업/아웃 혹은 샤딩을 하는 방식으로 확장하여 대응 가능하다고 한다.

  필자 또한 분산 트랜잭션에 관한 어려움과 현재 지식의 부족함을 고려하여 단일 저장소(RDBMS, MySQL)을 사용하여 이벤트(아웃박스)를 저장하기로 하였다.

이벤트의 상태

  핵심 로직을 실행하는 애플리케이션에서는 핵심 로직 발생에 대한 이벤트를 발행해야 한다. 즉, 아래와 같은 2가지 역할은 수행을 해야한다.

• 이벤트의 저장
• 이벤트의 발행

  따라서, 이벤트는 막 생성되어 저장된 이벤트 / 메시지 브로커로 발행된 이벤트 / 메시지 브로커로 발행에 실패한 이벤트 등 다양한 상태가 존재하게 된다.

  가장 중요한 것은 이벤트의 발행 여부이다.

  EDA를 통하여 모듈 간 강결합성을 낮춰 관심사의 분리, 장애 격리의 장점을 얻었다.
  비즈니스 도메인의 부가 로직에 대한 관심은 낮아졌지만, 부가 로직 실행이 보장되어야하기에 이벤트 발행 책임이 생긴다.

  이벤트의 상태는 이벤트 발행/재발행에 있어 필수적으로 필요한 속성이다.
  메시지 릴레이(Message Relay)를 통해서 이벤트 아웃박스를 조회하기 위해서는 아직 발행되지 않은 이벤트들만 조회해야 한다. 또한, 이미 발행된 이벤트는 삭제를 하고, 발행에 실패한 이벤트는 재발행을 시도하는 등의 처리를 위해서 이벤트 아웃박스의 상태가 필요하다.

트랜잭션 시점(Before/After Commit)에 따른 이벤트 처리 분리

  트랜잭션 아웃박스 패턴의 핵심 개념 중 하나는 비즈니스 로직과 이벤트 기록을 하나의 트랜잭션으로 묶어 원자적 수행을 하는 것이다.

  SpringEvent를 사용할 경우 커밋, 롤백 등 트랜잭션 전/후 등의 시점에서 Event를 처리할 수 있는 어노테이션 @TransactionalEventListener를 제공한다.

  이 중 ‘커밋 이전 시점(Before Commit)’과 ‘커밋 이후 시점(After Commit)’에서 이벤트를 처리할 수 있다.

• 이벤트 기록: TransactionPhase.BEFORE_COMMIT
• 이벤트 발행: TransactionPhase.AFTER_COMMIT

  이벤트 발행의 경우에는 메시지 릴레이를 통해서 조회를 하여 발행하지만, SpringEvent에서 제공하는 After Commit 시점의 이벤트 리스너를 통해서 메시지를 바로 발행할 수도 있다.

  TransactionPhase.BEFORE_COMMIT 시점에서는 이벤트를 이벤트 아웃박스 저장소에 기록(저장)한다.
  만약 이벤트 저장소에 기록이 실패하게 되면 트랜잭션 전체가 실패하게 되며 롤백되게 된다.

  TransactionPhase.AFTER_COMMIT 시점에서는 이벤트를 메시지 브로커로 발행한다.
  After Commit 시점에서 이벤트를 발행하지 않아도 무방하다. SpringEvent는 @TransactionalEventListener를 통해서 트랜잭션 시점에 따라 제어가 가능하지만, 다른 프레임워크를 사용하거나 SpringEvent를 이용하지 않는다면 트랜잭션 시점에 따른 처리가 불가능할 수도 있기 때문에 Microservice Architecture - Pattern: Transactional outbox에서도 메시지 릴레이를 통한 주기적 폴링 방식으로 이벤트 아웃박스를 조회해 발행하는 방식을 소개한다.

  After Commit 시점에서 이벤트 리스너를 통해 이벤트 발행 로직을 바로 수행한다면, 메시지 릴레이에서 조회할 이벤트 아웃박스의 양이 줄어들게 된다. 또한, 이벤트 발행이 실패하더라도 이미 저장소에 이벤트 아웃박스가 저장되어 있기 때문에 향후 폴링 시점에서 재발행이 가능하다. 따라서, After Commit 시점에서 이벤트 발행 로직을 수행하는 방향으로 설계하였다.

이벤트 폴링

  트랜잭션 아웃박스 패턴에서 데이터베이스에 저장된 이벤트 아웃박스를 주기적으로 조회(Polling)하여 메시지 브로커로 이벤트를 발행한다.

  메시지 릴레이(Message Relay)는 이벤트 아웃박스의 주기적인 조회를 담당하는 이벤트 폴러(Event Poller)와 외부로 이벤트를 발행하는 외부 이벤트 발행기(External Event Publisher)가 합쳐진 개념이다.

  필자는 실제 구현 상에 있어서 이벤트 폴러와 외부 이벤트 발행기를 분리하여 구현하였다.

  이벤트 폴러의 경우에는 상태에 따라 조회해야할 쿼리가 다양하다. 또한, 외부 이벤트 발행기도 After Commit 시점에서도 동시에 사용될 수 있다. 따라서, 각 클래스간 응집도와 재사용성을 고려하여 분리하여 구현하였다.

  이에 관한 자세한 내용은 차후 포스팅에서 다룰 예정이다.

Summary

  MSA 환경에서 Event Driven Architecture를 사용한다면 메시지 브로커를 통해 다른 모듈로 이벤트를 전달하게 된다. 이를 통해 핵심 비즈니스 도메인에서 부가 로직과의 결합도를 낮추어 관심사를 분리할 수 있다.

  어떤 부가 로직이 실행될지에는 관심을 가질 필요는 없지만, 부가 로직이 실행되기 위해서는 메시지 발행이 보장되어야 한다.

  이때, 메시지 유실과 핵심 로직과 메시지 발행의 원자적 실행을 위해 트랜잭션 아웃박스 패턴(Transactional Outbox Pattern)을 사용하게 된다.

  트랜잭션 아웃박스 패턴의 핵심은 이벤트를 저장소에 저장하는 것이다.

  이벤트를 저장소에 저장하기때문에 메시지 유실 문제를 해결하고, 비즈니스 도메인 수정과 메시지 발행을 하나의 트랜잭션으로 묶어 원자적인 연산을 수행하도록 한다.

  이벤트는 아웃박스(Outbox)의 형태로 저장된다. 아웃박스는 ‘보낼 편지함’이라는 사전적 의미처럼 발행을 위해 임시로 저장된 이벤트들을 나타내는 개념이다. 아웃박스는 발행/재발행 등을 처리하기 위하여 상태를 기록해야 한다.

  SpringEvent는 @TransactionalEventListener를 통해 트랜잭션 시점에 따라 이벤트를 처리할 수 있다. BEFORE_COMMIT 시점에서는 이벤트를 아웃박스의 형태로 저장하며, AFTER_COMMIT 시점에서는 이벤트를 발행한다.

  이벤트 폴러(Event Poller)는 주기적으로 저장소에 저장된 이벤트 아웃박스를 조회하여 외부 이벤트 발행기(External Event Publisher)를 통하여 외부 메시지 브로커로 이벤트를 발행한다.

  이를 통하여 이벤트 기반 아키텍처에서 관심사의 분리, 결합도 감소와 전송 보장이라는 장점을 동시에 가질 수 있게 된다.

  향후 포스팅에서 트랜잭션 아웃박스 패턴을 적용시킨 과정을 단계적으로 나누어 서술할 예정이다.

# Reference

• Microservice Architecture - Pattern: Transactional outbox
• 우아한 기술 블로그 - 회원시스템 이벤트기반 아키텍처 구축하기

  최근 ‘여행 기록 관리 플랫폼’ 프로젝트를 진행하며 이메일 발송 보장을 위하여 Transactional Outbox Pattern을 적용하기로 하였다. 이를 위하여 이벤트를 발송하는 일부 로직에 Spring Event 기반 구조를 도입하게 되었다.

  이벤트를 아웃박스로 변환하여 저장할 때, 이벤트를 식별하기 위한 식별자가 필요하다.

  이벤트를 식별하는 것 뿐만 아니라 처리된 이벤트에 대한 로그를 기록, 카프카로 이벤트 메시지 발행 시 처리 등을 위하여 이벤트를 구분하기 위한 식별자가 필요하다.

  이벤트 기반 아키텍처(EDA)는 기본적으로 도메인 별로 서비스와 최적화된 데이터베이스를 별도로 가진다. 이러한 분산 환경에서 Auto-Increment 식별자는 데이터를 고유하게 식별하는데 어려움이 존재한다. 따라서, 순차적 PK(Sequential PK)가 아닌 비순차적 PK(Non-Sequential PK)를 사용하게 된다.

  해당 프로젝트는 ‘모놀리식 단일 모듈’ 프로젝트이지만 차후 확장성을 고려하고, 애플리케이션 단에서 이벤트 아웃박스를 식별할 수 있는 eventId 컬럼 값을 생성하기로 설계하였기에 Non-Sequential PK를 사용하기로 하였다.

정확하게는 Non-Sequential한 값을 PK(id)로 사용하는 것은 아니다.
아웃박스의 PK는 Auto-Increment가 가능한 값을 사용하고, 아웃박스 별로 식별 가능한 별도의 컬럼인 eventId를 두어 사용한다.
해당 방식을 사용한 이유는 포스팅 후반부에 설명한다.

  eventId를 애플리케이션 레벨에서 생성하기로 한 이유 등은 향후 Transactional Outbox Pattern에 관한 포스팅에서 자세하게 다룰 예정이다.

  이번 포스팅에서 중점적으로 다룰 주제는 Non-Sequential PK로 사용되는 값인 UUID와 ULID의 차이와 레코드 삽입, 조회 시 발생하는 성능 비교이다.

  프로젝트에서는 ‘비밀번호 초기화 요청’과 ‘이메일 인증 코드 요청’ 기능 수행 후 이벤트를 발행하여 메일 발송이 이루어지도록 설계하였다.

  이는 지속적으로 발생하는 이벤트가 아닌 일회성 이벤트이기 때문에, 아웃박스에 저장되는 데이터의 양이 매우 많을 것이라고 보기는 어렵다. 다만 향후 MSA 및 EDA 전환 가능성을 고려하여, 대규모 프로젝트 환경을 가정하고 이벤트 및 아웃박스 구조를 고민하였다.

  eventId 컬럼의 식별자로 처음에 고려한 것은 UUID였다.
  그러나 UUID를 키로 사용할 경우 성능상 문제가 발생할 수 있다는 내용을 이전에 접한 기억이 있어, 이벤트가 대량으로 발행/저장되는 환경에서는 적절하지 않을 수 있다고 판단하였다. 이에 따라 UUID를 식별자로 사용할 때 발생할 수 있는 문제점에 대해 보다 자세히 조사하였다.

키의 특성이 데이터베이스 성능에 영향을 미칠 수 있는 요소

  우선, UUID나 ULID 등에 대해 알아보기 전에 키의 특성이 데이터베이스 성능에 영향을 미칠 수 있는 요소가 무엇이 있는지 정리해보고자 한다.

• 키의 크기 (Size of Key)
• 키의 순차성 (Sequentiality of Key)

  이번 테스트를 통해 위 2가지의 요소가 향후 UUID와 ULID의 성능 차이를 일으킨다는 것을 알게되었다.

  정확하게는 키의 크기는 작을수록, 키는 순차적일 수록 좋다고 할 수 있다. 자세하게는 위 2가지 요소를 기반으로 하여 아래의 특징들에서 성능 차이가 발생하게 된다.

1. 키도 곧 특정 레코드의 데이터이다.
2. 인덱스는 B+ Tree 구조를 가지며, 레코드 삽입 시 재배치가 이루어진다.
3. 페이지에 행(row)가 거의 다 찰 경우 페이지 분할(Page Split)을 통하여 추가적인 페이지를 확보한다.
4. RandomID를 사용할 경우 Cache Miss가 발생할 확률이 높다.

1. 키도 곧 특정 레코드의 데이터이다.

  말 그대로, 키도 곧 데이터이다.

  키의 길이가 길수록 한 행(row)의 크기가 커지기 때문에 페이지 내 들어갈 수 있는 데이터(row)의 갯수가 줄어든다.

2. 인덱스는 B+ Tree 구조를 가지며, 레코드 삽입 시 재배치가 이루어진다.

  인덱스는 데이터를 빠르게 찾기 위해 (키, 값) 쌍으로 구성된 B+ Tree 자료구조이다.

  MySQL에서는 주 인덱스와 보조 인덱스가 존재한다.
  주 인덱스는 클러스터링 인덱스로 키를 PK, 값을 레코드 전체로 가지는 인덱스이다. 즉, 테이블에 데이터 삽입 시 PK를 통해 값이 삽입된다.
  보조 인덱스는 논-클러스터링 인덱스로 키를 특정 컬럼, 값을 PK로 가지는 인덱스이다. 즉, PK를 제외하고 다른 컬럼들을 통해 사용자가 직접 지정하여 생성된 인덱스를 의미한다.

  인덱스는 B+ Tree의 키를 기준으로 정렬이 되어있으며, 리프노드들은 페이지이다. 주 인덱스는 각 페이지마다 레코드들이 저장되어있고, 보조 인덱스는 (키, PK) 쌍이 저장되어 있다.

  위 그림은 MySQL이 기본적으로 사용하는 데이터베이스 엔진인 InnoDB에서 사용하는 B+ Tree 구조를 나타낸 그림이다.

  리프노드는 실제 레코드가 담긴 페이지로 구성되어 있으며, 키를 기준으로 정렬되어 있는 것을 알 수 있다.

  즉, 새로운 데이터가 삽입이 되며 정렬된 키 순서를 유지하며 B+ Tree 구조를 만족하기 위해 재배치가 이루어진다.

  각 페이지가 꽉 찰 때까지는 해당 페이지 내에서 레코드의 정렬 순서를 변경하는 등의 방법으로 재배치를 수행하게 된다.

  그러나, 페이지가 꽉찬 상태에서 레코드가 삽입된 경우에는 페이지 분할이 필요하다.

3. 페이지에 행(row)가 거의 다 찰 경우 페이지 분할(Page Split)을 통하여 추가적인 페이지를 확보한다.

  해당 특징이 레코드 삽입 시 UUID와 ULID의 성능 차이를 불러일으키는 결정적 요인이다.

  페이지가 가득찬 상태에서 추가 레코드가 삽입될 경우, 키의 순차 여부에 따라 페이지가 분할되는 방식에서 차이가 발생한다.

  MySQL 공식 문서에 따르면 InnoDB의 인덱스는 추가 레코드가 삽입 되었을 때, 페이지 내 1/16 정도의 공간은 남겨둔다고 한다.

  따라서 결과적으로 생성되는 인덱스 페이지는 순차 삽입의 경우에는 대부분 15/16 만큼 채워져있으며, 랜덤 삽입의 경우에는 1/2(=50%)에서 15/16까지 채워져있다고 한다. 여기서 순차 삽입과 랜덤 삽입 시 결과 인덱스 페이지가 채워진 비율이 다르다는 점이 두 방식의 페이지 분할(Page Split) 기법도 다르다는 것을 나타낸다.

  페이지 분할 방식의 차이는 데이터베이스 오픈소스 소프트웨어 프로젝트 개발 회사인 Percona의 기술 블로그 포스트 “The Impacts of Fragmentation in MySQL”에서 확인할 수 있었다.

  해당 포스트에 따르면, InnoDB 엔진은 레코드가 삽입되었을 때 매우 영리한 방법으로 동작한다고 한다. 새로운 페이지(empty page)를 만들지, 페이지를 분할(page split)할지는 삽입되는 레코드에 따라 다르다고 한다.

  순차 삽입의 경우에는 기존의 페이지를 분할하는 것이 아닌 새로운 페이지를 만들어 해당 페이지에 레코드를 삽입한다고 한다. Auto Increment 키를 사용하는 것이 성능상 큰 이점을 가지고 온다는 것이다.

  그러나, 랜덤 삽입의 경우에는 향후 삽입될 레코드가 어느 위치에 삽입이 될 지 모르기 때문에 기존 페이지도 충분한 여유 공간을 확보해야한다. 따라서, 페이지 분할 기법을 이용하여 기존 페이지의 레코드의 절반을 새로 생성된 페이지에 옮긴다. 따라서, 각 새로 생성되는 페이지에 절반의 레코드를 옮기기 때문에 결과적으로 1/2(=50%) ~ 15/16까지 인덱스 페이지 내 레코드가 채워지게 되는 것이다.

  페이지 분할(Page Split) 자체도 되게 비용이 큰 무거운 연산이며, 레코드의 크기가 클 수록 페이지는 더욱 빨리 채워질 것이기에 페이지 분할도 많이 발생할 것이다.

4. RandomID를 사용할 경우 Cache Miss가 발생할 확률이 높다.

  MySQL은 레코드 조회 시 해당 레코드만 메모리로 로드하는 것이 아니라, 해당 레코드가 포함된 페이지를 읽어와 InnoDB Buffer Pool에 적재한다.

  즉, 페이지에 담긴 레코드의 수가 많을 수록 더 많은 레코드들이 메모리의 Buffer Pool에 올라올 수 있게 되는 것이다.

  RandomID의 경우에는 비슷한 시간대에 생성된 레코드이더라도 위치한 페이지가 다를 가능성이 높기 때문에 Buffer Pool에 해당 페이지가 존재하지 않는 Cache Miss가 발생할 확률이 높다. 즉, 이후 레코드를 또 조회하기 위해서는 해당 레코드를 포함한 페이지 로드 Disk I/O 작업이 빈번하게 이루어지게 되는 것이다.

  키의 순차성뿐만이 아니라 키의 크기가 클수록 Cache Miss의 발생 확률이 증가한다. 키도 레코드의 데이터 중 하나이기 때문에 키의 크기가 클수록 레코드의 크기도 증가하게 된다. 레코드의 크기가 증가하면 페이지 내 적재될 수 있는 레코드의 수는 줄어들게 된다. 따라서, Cache Miss 발생 확률이 증가하게 되는 것이다.

UUID vs ULID

  키의 크기와 키의 순차성에 따라 데이터베이스 성능에 영향을 미치게 된다. 그렇다면 UUID와 ULID는 어떠한 차이가 있는지 알아보자.

UUID

  UUID(Universally Unique Identifier)는 128-bit의 고유 식별자이다. 중앙 시스템에서 ID를 발급하는 형식이 아니기에 빠르고 간단하게 ID를 생성할 수 있는 방법이다. 필자가 애플리케이션 레벨에서 이벤트를 구분하기 위한 eventId 값을 만들기 위해 가장 먼저 생각난 방법이 UUID이다.

  UUID는 8-4-4-4-12 자리로 구성되어있으며 16진수로 표현한다. 특히, 3번째 필드의 첫번째 자리는 버전 정보를 나타낸다. 하이픈(-)까지 포함하면 총 36글자이다.

  Java에서는 java.util 패키지에서 UUID 클래스를 제공한다. UUID는 다양한 버전이 있으면 해당 클래스는 UUID v4 기준이다.

• UUID v1: 시간 + MAC 주소
• UUID v2: 시간 + POSIX
• UUID v3: 고정된 이름 + Namespace ⇒ MD5 해시
• UUID v4: 무작위 랜덤값(버전 정보 제외)
• UUID v5: 고정된 이름 + Namespace ⇒ SHA-1 해시
• UUID v6: 시간(정렬하기 좋게 재배치) + MAC
• UUID v7: Unix 시간 + 랜덤값

  Java에서 기본적으로 제공하는 UUID의 경우에는 완전 무작위값이므로 충돌 가능성이 매우 낮지만, 이를 키로 사용할 경우에 앞서 보았던 데이터베이스 성능 문제가 발생할 수 있다. 따라서, 데이터베이스의 키 값으로 UUID 사용을 고려한다면 UUID v4 보다는 타임스탬프(시간) 값을 기반으로 하는 다른 버전의 UUID를 사용하는 것이 도움이 될 것이다.

ULID

  ULID(Universally Unique Lexicographically Sortable Identifier)는 이름 그대로 사전적으로 정렬 가능한 범용 고유 식별자를 의미한다. 여기서, 사전적으로 정렬 가능하다라는 뜻은 ASCII 코드를 기준으로 문자의 크기를 비교하여 정렬하는 것을 나타낸다.

01EX8Y7M8M DVX3M3EQG69EEMJW
01EX8Y7M8M DVX3M3EQG69EEMJX
01EX8Y7M8M DVX3M3EQG69EEMJY
01EX8Y7M8M DVX3M3EQG69EEMJZ
01EX8Y7M8M DVX3M3EQG69EEMK0
01EX8Y7M8M DVX3M3EQG69EEMK1
01EX8Y7M8M DVX3M3EQG69EEMK2
01EX8Y7M8M DVX3M3EQG69EEMK3

01EX8Y7M8N 1G30CYF2PJR23J2J < millisecond changed
01EX8Y7M8N 1G30CYF2PJR23J2K
01EX8Y7M8N 1G30CYF2PJR23J2M
01EX8Y7M8N 1G30CYF2PJR23J2N
01EX8Y7M8N 1G30CYF2PJR23J2P
01EX8Y7M8N 1G30CYF2PJR23J2Q
01EX8Y7M8N 1G30CYF2PJR23J2R
01EX8Y7M8N 1G30CYF2PJR23J2S
         ^                ^
|--------|----------------|
   time      randomness
   (48)         (80)

  ULID는 Timestamp 48-bit + Randomess 80-bit로 구성되어 총 128-bit로 구성된 고유 식별자이다.

  UUID와 동일하게 128-bit이지만, ULID는 32진수를 쓰기 때문에 총 26글자로 구성된다.

  ULID의 Timestamp는 millisecond까지 감지 하기 때문에 1 ms가 지나면 앞 48-bit의 최하위 비트를 1 증가시키는 형식이다. 또한, 동일한 시간대이더라도 하위 80-bit가 랜덤으로 주어진다. 즉, 동일한 시점이더라도 2^80 가지 경우의 수가 존재한다.

  하지만, 난수라고해서 무조건 신뢰할 수 없기 때문에 동일한 시점에 생성된 ULID는 초기 랜덤 80-bit를 기준으로 1씩 증가하며 생성되기 때문에 충돌 가능성이 거의 없도록 한다.

UUID와 ULID의 차이

  이 시점에서 앞서 보았던 데이터베이스의 성능 차이를 발생시키는 요소에 기반하여 UUID(v4 기준)와 ULID를 비교하면 다음과 같다.

• UUID는 36글자, ULID는 26글자로 구성된다.
• UUID는 완전한 무작위 값(=비순차적), ULID는 타임스탬프 기반 값(=순차적)으로 구성된다.

  이러한 사실만 놓고 보더라도 InnoDB 엔진을 사용하는 MySQL에 레코드의 식별자로 UUID보다 ULID를 사용할 경우에 성능적으로 이점이 있다는 것을 예측할 수 있다.

  그러나, 학부생 수준에서 진행하는 프로젝트에서는 대규모 데이터가 생성되고 저장되는 일이 잘 없기 때문에, 구체적인 수치를 눈으로 확인하고 비교하기 위해 Spring과 MySQL을 통한 실험을 진행하였다.

UUID vs ULID 성능 비교 테스트

  해당 테스트는 식별자로 사용하는 값에 따른 인덱스 재배치 및 페이지 분할 발생 빈도수 차이 등을 확인하는 것을 목적으로 하고 있기에 레코드 삽입에 중점이 맞추어져있다.

  테스트 환경은 다음과 같다.

• Framework: SpringBoot 4.0.1
• Database: MySQL 9.2.0
• CPU: Apple M1
• RAM: 16GB
• Library: ULID Creator 5.2.3

  각 테이블은 UUID와 ULID를 PK로 가지며, 각 식별자의 정확한 길이에 따른 비교를 위해 varchar 타입으로 구성하였다.

mysql> describe uuid_table;
+-------+--------------+------+-----+---------+-------+
| Field | Type         | Null | Key | Default | Extra |
+-------+--------------+------+-----+---------+-------+
| id    | varchar(36)  | NO   | PRI | NULL    |       |
| data  | varchar(255) | YES  |     | NULL    |       |
+-------+--------------+------+-----+---------+-------+

mysql> describe ulid_table;
+-------+--------------+------+-----+---------+-------+
| Field | Type         | Null | Key | Default | Extra |
+-------+--------------+------+-----+---------+-------+
| id    | varchar(26)  | NO   | PRI | NULL    |       |
| data  | varchar(255) | YES  |     | NULL    |       |
+-------+--------------+------+-----+---------+-------+

  각 테이블명은 uuid_table과 ulid_table이다.

  테이블 생성 후 초기 크기는 다음과 같다.

mysql> SELECT TABLE_NAME, DATA_LENGTH AS 'Pure_Data_Size_Bytes', ROUND(DATA_LENGTH / 1024, 2) AS 'Pure_Data_Size_KB', TABLE_ROWS AS 'Total_Rows' 
FROM information_schema.TABLES
WHERE TABLE_NAME LIKE '%uuid_table%' OR TABLE_NAME LIKE '%ulid_table%';

+------------+----------------------+-------------------+------------+
| TABLE_NAME | Pure_Data_Size_Bytes | Pure_Data_Size_KB | Total_Rows |
+------------+----------------------+-------------------+------------+
| ulid_table |                16384 |                16 |          0 |
| uuid_table |                16384 |                16 |          0 |
+------------+----------------------+-------------------+------------+

  각 테스트는 SpringBoot의 테스트 코드를 통해 이루어졌으며, JPA 사용으로 불필요한 오버헤드 방지를 위해 jdbcTemplate 기반으로 코드를 작성하였다.

@SpringBootTest
public class IndexComparisonTest {
    private final JdbcTemplate jdbcTemplate;
    private final int RECORD_SIZE = 100_000;
    
    @Autowired
    public IndexComparisonTest(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }
    
    @Test
    @DisplayName("UUID를 PK로 하는 테이블 삽입")
    public void insertUUID() {
        String sql = "INSERT INTO uuid_table (id, data) VALUES (?, 'data')";
        
        StopWatch st = new StopWatch();
        st.start("UUID");
        for (int i = 0; i < RECORD_SIZE; i++) {
            jdbcTemplate.update(sql, UUID.randomUUID().toString());
        }
        st.stop();
        System.out.println(st.prettyPrint());
    }

    @Test
    @DisplayName("ULID를 PK로 하는 테이블 삽입")
    public void insertULID() {
        String sql = "INSERT INTO ulid_table (id, data) VALUES (?, 'data')";
        
        StopWatch st = new StopWatch();
        st.start("ULID");
        for (int i = 0; i < RECORD_SIZE; i++) {
            jdbcTemplate.update(sql, UlidCreator.getUlid().toString());
        }
        st.stop();
        System.out.println(st.prettyPrint());
    }
}

  우선, 삽입할 레코드의 크기를 100,000개로 설정한 뒤 삽입 테스트를 진행하였다.

100,000개 레코드 삽입

(1) 삽입 소요 시간

  삽입 소요 시간에서는 근소한 차이를 보이지만 UUID가 조금 더 오래걸리는 것을 알 수 있다. 페이지 분할이나 인덱스 재배치 등으로 인해 발생하는 오버헤드로 인한 차이일 것이다. 이는 데이터의 수가 많을수록 더욱 극명하게 나타날 것이다.

(2) 페이지 분할 횟수

# uuid_table에 레코드 삽입 후
mysql> select name, count from INFORMATION_SCHEMA.INNODB_METRICS where name like 'index_page%';
+-----------------------------+-------+
| name                        | count |
+-----------------------------+-------+
| index_page_splits           |   564 |
| index_page_merge_attempts   |     0 |
| index_page_merge_successful |     0 |
| index_page_reorg_attempts   |     0 |
| index_page_reorg_successful |     0 |
| index_page_discards         |     0 |
+-----------------------------+-------+

# ulid_talbe에 레코드 삽입 후
mysql> select name, count from INFORMATION_SCHEMA.INNODB_METRICS where name like 'index_page%';
+-----------------------------+-------+
| name                        | count |
+-----------------------------+-------+
| index_page_splits           |   341 |
| index_page_merge_attempts   |     0 |
| index_page_merge_successful |     0 |
| index_page_reorg_attempts   |     0 |
| index_page_reorg_successful |     0 |
| index_page_discards         |     0 |
+-----------------------------+-------+

  페이지 분할 횟수에서는 100,000개의 레코드임에도 차이가 뚜렷하게 나타났다.

  UUID가 ULID보다 페이지 분할 횟수가 많은 이유는 키의 길이가 더 길고, 랜덤(비순차) 삽입이 이루어지기 때문에 기존 페이지와 새로 생성되는 페이지에 1/2씩 레코드를 나눠가지게 분할이 되기 때문임을 알 수 있다.

(3) 인덱스로 인해 할당된 페이지 수

mysql> SELECT
     database_name,
     table_name,
     index_name,
     stat_value AS 'Total_Pages',
     stat_description AS 'Description'
 FROM mysql.innodb_index_stats
 WHERE (table_name LIKE '%uuid_table%' OR table_name LIKE '%ulid_table%')
   AND stat_name = 'size';

+---------------+------------+------------+-------------+------------------------------+
| database_name | table_name | index_name | Total_Pages | Description                  |
+---------------+------------+------------+-------------+------------------------------+
| index_test    | ulid_table | PRIMARY    |         353 | Number of pages in the index |
| index_test    | uuid_table | PRIMARY    |         675 | Number of pages in the index |
+---------------+------------+------------+-------------+------------------------------+

  주 키로 인해 생성된 클러스터링 인덱스에 할당된 총 페이지 수는 ulid_table은 353, uuid_table은 675 페이지가 생성된 것을 알 수 있다.

  100,000개의 레코드 삽입 시 생성된 페이지 수는 약 1.9배 차이라는 것을 알 수 있다. 이또한, 레코드가 더욱 많아질수록 차이가 클 것이다.

(4) 테이블 크기

mysql> SELECT TABLE_NAME, DATA_LENGTH AS 'Pure_Data_Size_Bytes', ROUND(DATA_LENGTH / 1024 / 1024, 2) AS 'Pure_Data_Size_MiB'
FROM information_schema.TABLES 
WHERE TABLE_NAME LIKE '%uuid_table%' OR TABLE_NAME LIKE '%ulid_table%';

+------------+----------------------+-------------------+
| TABLE_NAME | Pure_Data_Size_Bytes | Pure_Data_Size_MiB |
+------------+----------------------+-------------------+
| ulid_table |              5783552 |              5.52 |
| uuid_table |             11059200 |             10.55 |
+------------+----------------------+-------------------+

  테이블 자체 크기는 ulid_table는 5.78MB, uuid_table은 11MB이다.

  이또한 약 2배 정도의 차이를 보이고 있다.

(5) 페이지 당 평균 레코드 수 및 Fill Factor

mysql> SELECT 
    SUBSTRING_INDEX(TABLE_NAME, '.', -1) AS 'Pure_Table_Name',
    COUNT(*) AS 'Pages_In_Buffer',
    ROUND(AVG(DATA_SIZE) / 16384 * 100, 1) AS 'Real_Fill_Factor_Percent',
    ROUND(AVG(NUMBER_RECORDS), 1) AS 'Avg_Records_Per_Page'
FROM information_schema.INNODB_BUFFER_PAGE
WHERE TABLE_NAME LIKE '%uuid_table%' OR TABLE_NAME LIKE '%ulid_table%'
AND INDEX_NAME = 'PRIMARY'
GROUP BY TABLE_NAME;
+-----------------+-----------------+--------------------------+----------------------+
| Pure_Table_Name | Pages_In_Buffer | Real_Fill_Factor_Percent | Avg_Records_Per_Page |
+-----------------+-----------------+--------------------------+----------------------+
| `ulid_table`    |             341 |                     91.5 |                294.3 |
| `uuid_table`    |             580 |                     64.5 |                173.4 |
+-----------------+-----------------+--------------------------+----------------------+

  페이지 당 할당된 레코드 수는 Buffer Pool에 올라온 페이지를 기준으로 확인한다.

  MySQL의 information_schema.INNODB_BUFFER_PAGE에서는 버퍼풀에 올라온 페이지에 대한 다양한 정보를 제공한다.

  이번 지표(Metric)에서 주요하게 확인할 것은 Fill Factor와 페이지 당 평균 레코드 수이다.

  Fill Factor의 경우에는 ulid_table은 평균적으로 91.5%의 비율, uuid_table은 평균적으로 64.5%의 비율로 레코드가 채워져있다. MySQL InnoDB 엔진 설명과 유사한 결과 양상을 보이고 있다.

  페이지 당 평균 레코드 수는 ulid_table은 평균적으로 294.3개, uuid_table은 평균적으로 173.4개의 레코드가 채워져있다. UUID가 키의 길이가 더욱 길기도 하며, 페이지 분할 시 Fill Factor의 차이로 인하여 페이지 당 평균 레코드수가 낮은 것을 확인할 수 있다.

  페이지 당 평균 레코드 수가 낮으므로 UUID 사용 시 Cache Miss 확률이 더 높을 것이다.

(6) I/O Write Request

mysql> SELECT
        table_name,
        rows_fetched,
        rows_inserted,
        io_read_requests,
        io_write_requests
    FROM sys.schema_table_statistics
    WHERE table_name LIKE '%uuid_table%' OR table_name LIKE '%ulid_table%';

+------------+--------------+---------------+------------------+-------------------+
| table_name | rows_fetched | rows_inserted | io_read_requests | io_write_requests |
+------------+--------------+---------------+------------------+-------------------+
| ulid_table |            0 |        100000 |                0 |               420 |
| uuid_table |            0 |        100000 |                0 |              2681 |
+------------+--------------+---------------+------------------+-------------------+

  다음은 테이블에 대한 레코드(row)를 조회, 삽입 횟수와 Read/Write Disk I/O를 나타낸 지표이다.

  두 테이블 모두 100,000개의 레코드가 삽입되었지만 io_write_requests의 경우에는 약 6배 이상의 극명한 차이를 보인다.

  UUID의 경우 앞서 말한 페이지 분할의 특성으로 인하여 기존 페이지 수정 + 새로 생긴 페이지 기록 + 인덱스 트리 구조 수정 등 I/O 작업이 다수 발생하며, 더욱 빈번하게 발생하므로 이러한 차이가 발생한다는 것을 알 수 있다.

300,000개 레코드 삽입

  테스트 코드에서 RECORD_SIZE = 300_000으로 수정하여 300,000개 레코드 삽입 시에 발생하는 성능 차이를 확인하였다. 이때, 기존 테이블은 삭제 후 새로 생성하여 테스트를 진행하였다.

(1) 삽입 소요 시간

  100,000개의 레코드 삽입 시에는 약 2초 정도의 차이가 있었지만, 300,000개의 레코드 삽입 시에는 UUID가 ULID보다 약 22초 이상 더 소요된 것을 확인할 수 있다.

(2) 페이지 분할 횟수

mysql> SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE name = 'index_page_splits';
+-------------------+-------+
| name              | count |
+-------------------+-------+
| index_page_splits |  1838 |
+-------------------+-------+

mysql> SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE name = 'index_page_splits';
+-------------------+-------+
| name              | count |
+-------------------+-------+
| index_page_splits |  1019 |
+-------------------+-------+

  100,000개의 레코드 삽입 시에는 페이지 분할 횟수 차이가 223인 반면, 300,000개 레코드 삽입 시에는 819회 차이가 나는 것을 확인할 수 있다. 이는 약 1.8배 정도의 차이이다.

(3) 인덱스로 인해 할당된 페이지 수

mysql> SELECT
    database_name,
    table_name,
    index_name,
    stat_value AS 'Total_Pages',
    stat_description AS 'Description'
FROM mysql.innodb_index_stats
WHERE (table_name LIKE '%uuid_table%' OR table_name LIKE '%ulid_table%')
AND stat_name = 'size';

+---------------+------------+------------+-------------+------------------------------+
| database_name | table_name | index_name | Total_Pages | Description                  |
+---------------+------------+------------+-------------+------------------------------+
| index_test    | ulid_table | PRIMARY    |         996 | Number of pages in the index |
| index_test    | uuid_table | PRIMARY    |        1897 | Number of pages in the index |
+---------------+------------+------------+-------------+------------------------------+

  할당된 페이지 수 또한 UUID가 ULID보다 901개의 페이지가 더 많이 생성된 것을 확인할 수 있다.

(4) 테이블 크기

mysql> SELECT TABLE_NAME, DATA_LENGTH AS 'Pure_Data_Size_Bytes', ROUND(DATA_LENGTH / 1024 / 1024, 2) AS 'Pure_Data_Size_MiB' FROM information_schema.TABLES  WHERE TABLE_NAME LIKE '%uuid_table%' OR TABLE_NAME LIKE '%ulid_table%';

+------------+----------------------+--------------------+
| TABLE_NAME | Pure_Data_Size_Bytes | Pure_Data_Size_MiB |
+------------+----------------------+--------------------+
| ulid_table |             17367040 |              16.56 |
| uuid_table |             33177600 |              31.64 |
+------------+----------------------+--------------------+

  테이블 크기의 경우에도 약 2배 정도의 차이를 보이고 있으며 단순 텍스트 데이터만 가진 레코드이더라도 극명한 차이를 보이고 있다는 것을 알 수 있다.

(5) 페이지 당 평균 레코드 수 및 Fill Factor

mysql> SELECT 
    SUBSTRING_INDEX(TABLE_NAME, '.', -1) AS 'Pure_Table_Name',
    COUNT(*) AS 'Pages_In_Buffer',
    ROUND(AVG(DATA_SIZE) / 16384 * 100, 1) AS 'Real_Fill_Factor_Percent',
    ROUND(AVG(NUMBER_RECORDS), 1) AS 'Avg_Records_Per_Page'
FROM information_schema.INNODB_BUFFER_PAGE
WHERE TABLE_NAME LIKE '%uuid_table%' OR TABLE_NAME LIKE '%ulid_table%'
  AND INDEX_NAME = 'PRIMARY'
  GROUP BY TABLE_NAME;
+-----------------+-----------------+--------------------------+----------------------+
| Pure_Table_Name | Pages_In_Buffer | Real_Fill_Factor_Percent | Avg_Records_Per_Page |
+-----------------+-----------------+--------------------------+----------------------+
| `uuid_table`    |            1755 |                     63.9 |                171.9 |
| `ulid_table`    |            1022 |                     91.6 |                294.5 |
+-----------------+-----------------+--------------------------+----------------------+

  Fill Factor의 경우에는 기존과 유사하게 ulid_table은 평균적으로 91.6%의 비율, uuid_table은 평균적으로 63.9%의 비율로 레코드가 채워진다는 것을 알 수 있다.

  페이지당 평균 레코드 수 또한 기존과 유사하게 ulid_table은 171.9개, uuid_table은 294.5개를 가진다.

  따라서, 삽입되는 레코드의 수가 증가하더라도 이러한 양상은 유지된다는 것을 알 수 있다.

(6) I/O Write Request

SELECT
    table_name,
    rows_fetched,
    rows_inserted,
    io_read_requests,
    io_write_requests
FROM sys.schema_table_statistics
WHERE table_name LIKE '%uuid_table%' OR table_name LIKE '%ulid_table%';

+------------+--------------+---------------+------------------+-------------------+
| table_name | rows_fetched | rows_inserted | io_read_requests | io_write_requests |
+------------+--------------+---------------+------------------+-------------------+
| uuid_table |            0 |        300000 |                0 |            106372 |
| ulid_table |            0 |        300000 |                0 |              1316 |
+------------+--------------+---------------+------------------+-------------------+

  I/O Write Request의 경우 UUID는 106,372회, ULID는 1,316회 발생하였으며 이는 약 80배의 차이이다.

  삽입되는 레코드 양이 커질수록 페이지 분할 및 인덱스 재배치 횟수도 커지기 때문에 극명한 차이를 보이게 된다.

  위의 테스트 과정을 통해 다음과 같은 사실들을 도출해낼 수 있었다.

• 키의 크기가 성능에 영향을 미친다.
  • 키가 클수록, 페이지 당 삽입 가능 레코드 수가 줄어들게 되어 성능이 저하한다.
• 키의 순차성이 성능에 영향을 미친다.
  • 순차키의 경우에는 페이지 분할 시, 새로운 페이지에 추가된 레코드를 삽입하게 된다.
  • 비순차키의 경우에는 페이지 분할 시, 새로운 페이지에 기존 페이지의 레코드 절반을 옮긴다.
  • 따라서, 비순차키 사용 시 페이지 분할과 인덱스 재배치가 빈번하게 일어나게 된다.
• UUID는 키의 크기가 상대적으로 크고, 랜덤(비순차)한 식별자이다.
• ULID는 키의 크기가 상대적으로 작고, 순차적인 식별자이다.
• 레코드의 양이 많을수록 데이터베이스 성능 차이가 극명하게 발생한다.

PK가 아닌 보조 인덱스로 사용할 경우

  앞서 서론 부분에 이벤트 아웃박스의 eventId 컬럼을 PK가 아닌 일반 컬럼으로 두고, 보조 인덱스로 사용한다고 하였다.

  보조인덱스로 사용한다면 레코드의 크기가 또 증가할텐데 이렇게 선택한 이유는 보조 인덱스는 논-클러스터링 인덱스이기 때문이다.

  실제 운영환경에서는 테이블의 크기가 큰 경우가 많다. 필자가 해당 기능을 위해 저장하는 이벤트 아웃박스도 createdAt, last_retried_at, payload 등 다양한 컬럼이 존재하며, payload의 경우에는 여러 이벤트마다 크기가 다르기 때문에 적당히 큰 크기로 설정해놓은 상태이다.

  이러한 상황에서 PK를 Non-Sequential PK로 사용할 경우, 인덱스 재배치가 이루어지는 과정에서 레코드 전체가 재배치가 발생하여 큰 오버헤드가 발생할 것이다.

  따라서, 레코드의 크기가 큰 경우에는 PK는 Auto Increment, 보조 인덱스는 키의 크기가 작고, Sequential한 값을 사용하는 것이 효율적일 것 이다.

  만약, PK는 Auto Increment한 값을 사용하고 보조 인덱스로 UUID와 ULID를 사용하는 테이블에 100,000개 레코드를 삽입한다면 결과는 다음과 같다.

// PK는 Auto Increment + 보조 인덱스는 ULID
+-------------------+-------+
| name              | count |
+-------------------+-------+
| index_page_splits |  1281 |
+-------------------+-------+

// PK는 Auto Increment + 보조 인덱스는 UUID
+-------------------+-------+
| name              | count |
+-------------------+-------+
| index_page_splits |  1783 |
+-------------------+-------+

+---------------+----------------------+------------+-------------+------------------------------+
| database_name | table_name           | index_name | Total_Pages | Description                  |
+---------------+----------------------+------------+-------------+------------------------------+
| index_test    | ulid_secondary_table | PRIMARY    |         995 | Number of pages in the index |
| index_test    | ulid_secondary_table | idx_ulid   |         289 | Number of pages in the index |
| index_test    | uuid_secondary_table | PRIMARY    |        1123 | Number of pages in the index |
| index_test    | uuid_secondary_table | idx_uuid   |         611 | Number of pages in the index |
+---------------+----------------------+------------+-------------+------------------------------+

mysql> SELECT TABLE_NAME, DATA_LENGTH AS 'Pure_Data_Size_Bytes', ROUND(DATA_LENGTH / 1024 / 1024, 2) AS 'Pure_Data_Size_MiB' FROM information_schema.TABLES  WHERE TABLE_NAME LIKE '%uuid_secondary_table%' OR TABLE_NAME LIKE '%ulid_secondary_table%';
+----------------------+----------------------+--------------------+
| TABLE_NAME           | Pure_Data_Size_Bytes | Pure_Data_Size_MiB |
+----------------------+----------------------+--------------------+
| ulid_secondary_table |             16302080 |              15.55 |
| uuid_secondary_table |             18399232 |              17.55 |
+----------------------+----------------------+--------------------+

+------------------------+------------+-----------------+--------------------------+----------------------+
| Pure_Table_Name        | INDEX_NAME | Pages_In_Buffer | Real_Fill_Factor_Percent | Avg_Records_Per_Page |
+------------------------+------------+-----------------+--------------------------+----------------------+
| `ulid_secondary_table` | PRIMARY    |            1035 |                     91.5 |                 97.6 |
| `ulid_secondary_table` | idx_ulid   |             251 |                     97.5 |                399.4 |
| `uuid_secondary_table` | PRIMARY    |            1091 |                     92.4 |                 92.7 |
| `uuid_secondary_table` | idx_uuid   |             506 |                     60.6 |                198.6 |
+------------------------+------------+-----------------+--------------------------+----------------------+

  위 결과에서 ULID를 보조 인덱스로 사용하는 경우, 주 인덱스와 보조 인덱스의 Fill Factor가 모두 90% 이상을 차지하고 있는 것을 알 수 있다.

  그러나, 보조 인덱스 자체도 인덱스를 형성해야하며 해당 컬럼들이 레코드 내 데이터이기 때문에 테이블의 크기가 거치고, 인덱스 자체가 2개이니 페이지 분할 횟수도 Non-Sequential한 값을 PK로 사용했을 때보다 증가한 것을 알 수 있다.

  따라서, ‘Auto Increment PK + 보조 인덱스는 UUID’ 조합은 테이블의 크기도 키우고, 페이지 분할 횟수와 인덱스 재배치 효율도 떨어지는 최악의 조합이라는 것을 알 수 있다.

  그렇다면 “ULID를 보조 인덱스로 사용해도 어차피 페이지 분할 횟수나 테이블 크기가 커지는데 이것이 과연 좋은가?”라는 의문을 가질 수 있다.

  그러나, 위 테스트는 레코드의 크기가 작은 경우이다. 레코드의 크기가 큰 상황을 가정하여 ULID를 PK로 사용하는 경우 vs ULID를 보조 인덱스로 사용하는 경우를 비교해보자.

mysql> describe heavy_ulid_pk;
+------------+---------------+------+-----+-------------------+-------------------+
| Field      | Type          | Null | Key | Default           | Extra             |
+------------+---------------+------+-----+-------------------+-------------------+
| id         | varchar(26)   | NO   | PRI | NULL              |                   |
| content    | varchar(2000) | YES  |     | NULL              |                   |
| created_at | timestamp     | YES  |     | CURRENT_TIMESTAMP | DEFAULT_GENERATED |
+------------+---------------+------+-----+-------------------+-------------------+

mysql> describe heavy_ulid_sec;
+------------+---------------+------+-----+-------------------+-------------------+
| Field      | Type          | Null | Key | Default           | Extra             |
+------------+---------------+------+-----+-------------------+-------------------+
| id         | bigint        | NO   | PRI | NULL              | auto_increment    |
| sub_id     | varchar(26)   | NO   | MUL | NULL              |                   |
| content    | varchar(2000) | YES  |     | NULL              |                   |
| created_at | timestamp     | YES  |     | CURRENT_TIMESTAMP | DEFAULT_GENERATED |
+------------+---------------+------+-----+-------------------+-------------------+

  레코드의 크기가 큰 2가지 테이블 heavy_ulid_pk와 heavy_ulid_sec 테이블을 만들어 100,000개 데이터를 삽입하여 결과를 비교하였다.

  인덱스 발생 횟수나 소요시간은 heavy_ulid_sec의 경우에는 추가적인 인덱스를 사용하기 때문에 큰 의미가 없어 해당 지표는 비교 대상에 포함하지 않았다.

+---------------+----------------+------------+------------------+------------------------------+
| database_name | table_name     | index_name | Total_Pages_Disk | Description                  |
+---------------+----------------+------------+------------------+------------------------------+
| index_test    | heavy_ulid_pk  | PRIMARY    |             3626 | Number of pages in the index |
| index_test    | heavy_ulid_sec | PRIMARY    |             3622 | Number of pages in the index |
| index_test    | heavy_ulid_sec | idx_ulid   |              161 | Number of pages in the index |
+---------------+----------------+------------+------------------+------------------------------+

+----------------+----------------------+--------------------+
| TABLE_NAME     | Pure_Data_Size_Bytes | Pure_Data_Size_MiB |
+----------------+----------------------+--------------------+
| heavy_ulid_pk  |             59408384 |              56.66 |
| heavy_ulid_sec |             59342848 |              56.59 |
+----------------+----------------------+--------------------+

+------------------+------------+-----------------+--------------------------+----------------------+
| Pure_Table_Name  | INDEX_NAME | Pages_In_Buffer | Real_Fill_Factor_Percent | Avg_Records_Per_Page |
+------------------+------------+-----------------+--------------------------+----------------------+
| `heavy_ulid_sec` | PRIMARY    |            3578 |                     90.5 |                 15.0 |
| `heavy_ulid_pk`  | PRIMARY    |            2755 |                     89.8 |                 15.2 |
| `heavy_ulid_sec` | idx_ulid   |             127 |                     96.4 |                394.7 |
+------------------+------------+-----------------+--------------------------+----------------------+

  테이블의 크기를 비교해보면, 레코드 내 다른 컬럼들의 크기가 크기 때문에 ULID 보조인덱스가 미치는 영향이 상대적으로 크지 않은 것을 알 수 있다.

  핵심적으로 차이를 보이는 부분은 ‘페이지 당 평균 레코드 수’와 ‘총 페이지 수’이다.

  주 인덱스의 Fill Factor는 Auto Increment한 값과 ULID 모두 순차성을 보이고 있기 때문에 약 90%의 좋은 결과를 보인다. 그러나, 레코드 자체의 크기가 크기 때문에 한 페이지당 평균 레코드 수가 15개로 매우 낮게 나온다.

  보조 인덱스 idx_ulid의 경우에는 Fill Factor가 약 96.4%, 페이지 당 평균 레코드 수가 약 400개이다. 보조 인덱스는 논-클러스터링 인덱스로 (인덱스 키, PK)를 쌍으로 가지고 있기 때문이다.

  즉, ULID를 PK로 사용하는 경우에는 한 페이지 당 레코드가 15개가 정도이기에, 레코드를 찾기 위해 3,626개의 페이지가 빈번하게 I/O 되는 일이 일어나게 된다.

  그러나, 보조 인덱스가 존재하는 경우에는 한 페이지 당 400개의 레코드 PK를 가지고 있고 페이지 수가 161개 뿐이기에 I/O에 드는 비용 차이가 확실히 나타나게 된다. 당연하게도, 레코드의 크기가 크기 때문에 주 인덱스에서는 I/O 작업도 무겁다고 할 수 있다.

  이를 기반으로 하여, 레코드 조회 시 성능 차이 여부를 확인해보고자 한다.

SET profiling = 1;

SELECT id FROM heavy_ulid_pk WHERE id LIKE '01J%';

SELECT sub_id FROM heavy_ulid_sec WHERE sub_id LIKE '01J%';

SHOW PROFILES;

mysql> SHOW PROFILES;
+----------+------------+------------------------------------------------------------+
| Query_ID | Duration   | Query                                                      |
+----------+------------+------------------------------------------------------------+
|        1 | 0.00638125 | SELECT id FROM heavy_ulid_pk WHERE id LIKE '01J%'          |
|        2 | 0.00113025 | SELECT sub_id FROM heavy_ulid_sec WHERE sub_id LIKE '01J%' |
+----------+------------+------------------------------------------------------------+

  ULID PK를 통하여 조회 쿼리를 실행한 결과는 6.38ms, ULID 보조인덱스를 통하여 조회 쿼리를 실행한 결과는 1.13ms가 소요된 것을 알 수 있다.

  레코드 조회 시 약 5.64배 조회 속도 차이가 발생하게 된다.

mysql> EXPLAIN SELECT id FROM heavy_ulid_pk WHERE id LIKE '01J%';
+----+-------------+---------------+------------+-------+---------------+---------+---------+------+------+----------+--------------------------+
| id | select_type | table         | partitions | type  | possible_keys | key     | key_len | ref  | rows | filtered | Extra                    |
+----+-------------+---------------+------------+-------+---------------+---------+---------+------+------+----------+--------------------------+
|  1 | SIMPLE      | heavy_ulid_pk | NULL       | range | PRIMARY       | PRIMARY | 106     | NULL |    1 |   100.00 | Using where; Using index |
+----+-------------+---------------+------------+-------+---------------+---------+---------+------+------+----------+--------------------------+

mysql> EXPLAIN SELECT sub_id FROM heavy_ulid_sec WHERE sub_id LIKE '01J%';
+----+-------------+----------------+------------+-------+---------------+----------+---------+------+------+----------+--------------------------+
| id | select_type | table          | partitions | type  | possible_keys | key      | key_len | ref  | rows | filtered | Extra                    |
+----+-------------+----------------+------------+-------+---------------+----------+---------+------+------+----------+--------------------------+
|  1 | SIMPLE      | heavy_ulid_sec | NULL       | range | idx_ulid      | idx_ulid | 106     | NULL |    1 |   100.00 | Using where; Using index |
+----+-------------+----------------+------------+-------+---------------+----------+---------+------+------+----------+--------------------------

  이는 당연하게도 조회 시 사용하는 인덱스가 보조 인덱스와 주 인덱스로 나뉘기 때문에 발생하는 차이이다.

  위 결과를 바탕으로 ULID를 PK로 사용하는 경우 vs ULID를 보조 인덱스로 사용하는 경우를 비교한다면 다음과 같다.

  레코드의 크기가 큰 실제 운영 상황에서는 하나의 레코드의 크기가 크기때문에 주 인덱스(클러스터링 인덱스)에서 페이지 당 레코드 갯수가 매우 적다.

  보조 인덱스의 경우에는 논-클러스터링 인덱스로 페이지 당 보유 레코드 수가 주 인덱스 대비 많다. 또한, 페이지의 갯수도 적다. 따라서, 레코드 조회 시 주 인덱스보다 보조 인덱스를 사용하는 것이 더욱 효율적인 성능을 발휘한다는 것을 알 수 있다.

  실제 운영 상황에는 레코드의 크기와 갯수, 보조 인덱스가 차지하는 비율 등을 잘 고려하여 적절한 방법을 선택해야 한다는 생각이 든다.

  이번 테스트를 통하여 InnoDB 엔진의 페이지 분할 동작 방식과 운영체제의 페이지 교체 등의 개념들이 실제 데이터 삽입, 조회 시 영향을 미치는 직접적인 사례를 확인할 수 있었다. 또한, 인덱스라는 개념이 단순히 데이터를 빠르게 찾아주는 수단이 아니라 어떻게 구성되어 있으며, 어떠한 방법으로 사용해야지 효율적으로 사용할 수 있는지 배울 수 있었다.

  여행 기록 관리 플랫폼 ‘여기가’ 프로젝트를 진행하며, 비밀번호 초기화 이메일 전송 기능 개발을 담당하게 되었다. 기능 개발에는 그리 오랜 시간이 소요되지는 않았지만, 구현을 완료한 뒤 에러 상황을 겪게 되며 한 가지 의문점이 들게 되었다.

  해당 에러는 SPF/DKIM 관련 에러였으나, 필자가 궁금증을 가지게 된 부분은 사용자에겐 정상 발송 응답이 되지만, 실제 이메일은 발송되지 않는 상황이다. 이메일 발송 부분은 @Async 어노테이션을 통한 비동기로 동작하도록 설정하였기에 해당 쓰레드에서 에러가 발생하더라도 이는 이메일 초기화 요청 핵심 비즈니스 로직이 동작하는 쓰레드에 영향을 미치지 못한다.

  해당 기능은 3분 동안 이메일 초기화 링크가 유효하며, 3분 이내 비밀번호 초기화 재요청 시 해당 요청은 거절된다. 즉, 이메일 발송이 실패한 경우에 사용자는 3분 간 아무것도 하지 못하는 상태가 되어버리는 문제가 발생하는 것이다.

  필자는 이러한 문제를 해결하기 위하여 비밀번호 초기화 이메일 발송이 보장되어야 한다는 생각을 가지게 되었고, 이에 대한 해결법을 찾아보게 되었다.

  결론적으로 ‘트랜잭션 아웃박스 패턴’이 해결법이라는 것을 알게 되었다. 트랜잭션 아웃박스 패턴을 통해 최소한 1번의 실행은 보장하는 ALO(At-Least Once)가 가능하다. 트랜잭션 아웃박스 패턴에 대한 더 자세한 내용과 적용은 차후 포스팅에서 다뤄보고자 한다.

  이번 포스팅에서 중점적으로 다루고자 하는 내용은 ‘이벤트’이다.

  트랜잭션 아웃박스 패턴에 대해 찾아보면, 대부분의 자료에서 이벤트 기반 아키텍처와 연관되어 있음을 알 수 있다. 처음에는 “왜 트랜잭션 아웃박스 패턴이 이벤트 기반 아키텍처와 연관되어있지?”라는 생각을 했었지만, 실제적으로는 “이벤트 기반 아키텍처에서 트랜잭션 아웃박스 패턴이 필요하다”는 것을 깨닫게 되었다.

  즉, 트랜잭션 아웃박스 패턴에 대해 알아보기 전에 ‘이벤트 기반 아키텍처’에 대한 이해가 필요하다.

이벤트 기반 아키텍처

  이벤트 기반 아키텍처에 대한 설명은 우아한형제들 2020 우아콘 김영한님의 ‘마이크로서비스 여행기’ 발표에서 이해할 수 있었다.

  해당 발표는 마이크로서비스를 중점으로한 내용이지만, 마이크로서비스가 동작하기 위해서 이벤트 기반 아키텍처가 도입되어야하는 이유를 알게 해주었다.

  메인 데이터베이스(루비)에 의존하던 레거시 시스템을 마이크로서비스로 전환하게 되는 과정에서 시스템 간 결합도를 최대한 분리하게 되는 과정을 엿볼 수 있었다.

  특정 도메인을 메인으로 관리하는 하나의 시스템에 조회나 검색 등 다른 시스템들이 의존하게 된다. 직접 API 호출을 통하여 데이터를 조회할 시, 연쇄적인 장애 전파 문제 및 사용자 트래픽이 전파되어 시스템 부하를 높이는 문제가 발생하게 된다. 따라서, 각 시스템이 역할에 맞는 데이터를 최적화하여 가지고 있을 수 있는 데이터베이스를 별도로 가지게 된다. CQRS에 기반하여 조회(Query) 시스템과 명령(Command) 시스템으로 구분되게 된다.

  결국, 조회 시스템의 경우 특정 도메인 시스템이 가지고 있는 테이블에서 자신이 사용자에게 제공할 최적화된 값들만 가지고 있어야 성능적인 면에서 유리하다. 즉, 특정 도메인의 일부 필요한 데이터를 가지고와서 해당 조회 시스템의 데이터베이스에 저장하게 된다.

  이 과정에서 데이터의 변경이 생기면, 해당 데이터를 참조하는 조회 시스템으로 변경 내용을 전파하기 위해서 가장 적절한 방식이 ‘이벤트 기반 아키텍처’이다.

  데이터 갱신 이벤트를 발행하고, 이를 이벤트 브로커(AWS SNS/SQS)를 통해 다른 시스템에 전파를 하게 된다. 해당 이벤트를 수신하게 되는 조회 시스템에서는 최신의 데이터를 받아오기 위해 이 시점에서 조회 API를 호출해 데이터를 갱신한다.

해당 방식은 Zero-Payload 방식을 사용하며, 최소한의 필요 데이터(ex. ID)만 전달해 데이터 정합성을 유지하는 방법이라 한다.

  여기서 이벤트를 사용하여 전달하는 것이 장애 분리, 확장성, 재시도, 전송 보장 등 다양한 방면에서 이점을 얻을 수 있게 된다.

장애 분리

  예를 들어 시스템 A와 B가 있다고 가정하면 다음과 같은 상황이 생길 수 있다.

• A 시스템의 핵심 비즈니스 로직 → B 시스템 API 직접 호출
• B 시스템에서 에러 발생
• A 시스템에도 에러가 전파되어 해당 트랜잭션이 실패

  위와 같이 장애가 전파되어 주관심사인 핵심 비즈니스 로직도 실패하는 상황이 생길 수가 있다.

  이벤트 기반 아키텍처를 사용하게 된다면, 핵심 비즈니스 로직에서는 이벤트만 발행하기 때문에 차후 이벤트를 활용하게 되는 타 시스템의 장애에 영향을 받지 않게 된다.

  이는, 마이크로서비스 아키텍처에서 타 시스템 간 장애 전파가 아닌 모놀리식 단일 모듈에서 객체 간 장애 전파 상황과도 비슷할 것이다.

@Async도 장애 분리일까?

  앞서, 비밀번호 초기화 메일 발송 과정에서 이메일 발송 로직 내 오류가 발생할 경우, 사용자에게 정상적인 응답이 가는 이유가 이메일 발송 로직이 @Async를 통한 비동기로 동작하기 때문이라 설명하였다.

  이메일 발송까지도 핵심 로직이긴 하지만, @Async를 사용한 부가 로직에서 에러가 발생한 경우 핵심 비즈니스 로직에서 이 에러가 전파되지 않는다는 상황만 두고보면 이는 “@Async를 사용한 경우에도 장애 분리가 된다고 볼 수 있나?”라는 의문이 들었다.

  결론적으로는 표면상 장애가 전파되지 않는다는 점은 맞긴하지만, 장애를 분리했다고 보기는 어렵다.

  왜냐하면, 이는 장애가 적절히 대응된 것이 아니라 별도의 쓰레드에서 동작하기 때문에 장애 인식 시점이 늦어지는 결과만 초래할 뿐이기 때문이다. 이는 필자의 개인적인 생각이라 정확한 해석은 아닐 수 있지만, 적어도 장애 분리가 @Async의 장점은 아니라고 말할 수 있을 것 같다.

이벤트

  앞선 내용에서 어느정도 이벤트에 대한 내용을 이해할 수 있었지만 해당 섹션에서 장점을 한 번 더 정리해보고자 한다.

  이벤트 기반 아키텍처에 관한 내용을 찾아보며 “그래서 이벤트가 왜 좋은건데?”라는 질문에 대한 답을 스스로 이해할 수 있었다.

  필자가 생각하는 이벤트를 사용함에 있어 주는 장점은 ‘느슨한 결합도 형성’과 ‘핵심 관심사와 부가 관심사의 분리’라고 생각한다. 앞서 보았던 ‘장애 격리’도 느슨한 결합도로 인해 생기는 이점이라 할 수 있다.

// AuthService
private final MemberService memberService;
private final MemberSignUpSnsSender memberSignUpSnsSender;
private final MemberSignUpEmailSender memberSignUpEmailSender;
private final MemberSignUpLogger memberSignUpLogger;

public void signUp(SignUpDto dto) {
    // 핵심 로직 (사용자 등록)
    Member member = memberService.create(dto.toUserEntity());

    // 부가 로직
    memberSignUpSnsSender.send(member);   // SNS 발송
    memberSignUpEmailSender.send(member); // Email 발송
    memberSignUpLogger.execute(member);   // 로그 기록
                                          // ...
}

  회원가입 기능을 예로 들면, 위 코드에서는 핵심 로직(사용자 등록)과 부가 로직이 같은 메서드 내에 위치한 것을 알 수 있다.

  우선, 해당 메서드 내 부가 로직이 함께하게 되어 코드의 응집도가 떨어진다는 문제점을 눈으로 확인할 수 있다. 해당 부가 로직들을 AuthService 내에서 실행해야하기 때문에 해당 의존성도 멤버 변수로 선언해야한다.

  의존성이 많아 클래스 자체가 비대해지며, AuthService 내 회원가입 메서드에서만 사용되는 부가 로직을 위해 테스트 시에 불필요한 Mock 객체를 다수 스터빙해야한다는 문제도 예상이 된다.

  또한, 부가 로직이 추가적으로 증가할 경우 AuthService 내 추가적인 의존성이 필요하며, 테스트 코드 재작성이 필요하다. 부가 로직 중 단 하나의 로직에서 장애가 발생할 경우 핵심 및 모든 부가 로직이 실패하게 되어 장애가 전파된다는 문제점도 존재한다.

// AuthService
private final MemberService memberSerivce;
private final EventPublisher eventPublisher;

public void signUp(SignUpDto dto) {
    // 핵심 로직 (사용자 등록)
    Member member = memberService.create(dto.toUserEntity());

    // 부가 로직 (이벤트 발행으로 통일)
    eventPublisher.publish(MemberSignUpEvent.from(member));
}

  이벤트를 적용한다면 핵심 비즈니스 로직의 코드가 간결해지며 응집도가 높아진다.

  AuthService가 의존하는 객체의 수도 줄게되어 멤버 회원가입과 연관된 부가 기능과의 강결합도를 해소할 수 있게 된다.

  회원가입 이후 추가적으로 실행해야할 부가 로직이 추가된 경우에도 AuthService 내 코드는 수정할 필요가 없으며, 이벤트를 소비하는 리스너를 추가 정의하여 사용하기만 하면 된다. 즉, 확장에 용이해진 것이다.

  Spring Event의 경우에는 트랜잭션의 범위를 제어할 수 있긴 하지만, 이벤트를 소비하는 부가 로직에서 적절한 예외 처리를 수행하는 등의 방법으로 장애 격리도 가능하다.

Spring Event

  이벤트 기반 아키텍처에서는 이벤트 발행자/소비자와 이벤트를 중개하는 이벤트 버스(브로커)가 필요하다.

  Kafka, RabbitMQ, AWS SNS/SQS 등 다양한 메시지 브로커들이 이벤트 버스(브로커) 역할을 수행한다. 이와 유사하게 Spring 내에 Spring Event가 존재한다.

  앞선 메시지 브로커들은 큐의 상태를 GUI로 확인하거나 명시적으로 큐의 상태를 확인하고 제어할 수 있지만, Spring Event의 경우에는 스프링 내부적으로 애플리케이션 레이어에서 이벤트를 제어하기 때문에 비슷하지만 다른 느낌을 받기도 한다.

  그렇다면 Spring Event를 사용하는 이유는 무엇일까?

  이는 '느슨한 결합'과 '트랜잭션 범위 제어'를 통한 비관심사를 효율적으로 처리하기 위함에 있다고 생각한다. 즉, 이벤트 기반 아키텍처를 사용하는 이유와 거의 동일한 것이다.

  시스템 간이 아닌, 애플리케이션 레이어에서도 주 관심사와 부가 관심사를 분리하여 느슨한 결합의 장점을 가져갈 수 있게 된다.

  그리고, Spring Event의 @EventListener를 확장한 @TransactionalEventListener 어노테이션에서는 트랜잭션 범위 제어에 관한 옵션을 설정할 수 있다.

• TransactionPhase.BEFORE_COMMIT
• TransactionPhase.AFTER_COMPLETION
• TransactionPhase.AFTER_COMMIT
• TransactionPhase.AFTER_ROLLBACK

  트랜잭션이 커밋되기 전/후, 트랙잭션 커밋 여부와 상관없이, 롤백되고 난 후 등 다양한 범위의 트랜잭션 제어가 가능하기 때문에 비관심사를 효율적으로 처리할 수 있다.

  트랜잭션 아웃박스 패턴의 경우 해당 트랙잭션에서 발생한 이벤트를 기록하는 것이 중요하다. 따라서, 트랜잭션이 커밋되기 이전에 해당 이벤트를 기록하여야 한다. 이후 트랜잭션 커밋이 완료된 시점에 기록된 이벤트를 활용한 추가 로직을 실행하도록하여 트랜잭션 범위에 따른 핵심/부가 관심사의 효율적인 제어가 가능하다.

  위 그림은 우아한형제들 기술블로그 ‘회원시스템 이벤트기반 아키텍처 구축하기’ 글에서 소개된 이벤트 기반 아키텍처이다.

  배달의 민족 시스템에서도 타 시스템에 이벤트 전달을 위해서는 AWS SNS/SQS를 사용하지만, Application 레이어 내부에서 Spring Event도 사용하는 것을 알 수 있다.

  해당 포스팅에서는 Spring Event를 사용하는 이유로 분산-비동기를 다룰 수 있으며 트랙잭션 범위 제어가 가능하다는 점을 활용하여, 비관심사이지만 시스템에서 반드시 해결해야하는 부가 로직을 도메인에 영향 없이 확장과 변경이 용이한 Spring Event을 적용하여 해결하였다는 것을 확인할 수 있었다.

Transactional Outbox Pattern

  비밀번호 초기화 메일 발송 기능의 전송을 보장하기 위한 방법을 찾아보던 중 가장 먼저 보았던 키워드는 Transactional Outbox Pattern이다.

  트랜잭션 아웃박스 패턴은 이벤트 기반 아키텍처에서 전송을 보장하기 위해 사용되는 방법이다. 따라서, 이번 포스팅에서 이벤트에 관한 내용을 우선 작성하게 되었다. 트랜잭션 아웃박스 패턴에 대한 내용은 차후 포스팅에서 더 자세하게 다루겠지만, 이번 포스팅에서 이벤트 기반 아키텍처에서 트랜잭션 아웃박스 패턴이 언급되는 이유를 간략하게 설명하고자 한다.

  이벤트 기반 로직에서 시스템 결합도를 낮추고, 장애 격리의 장점은 얻을 수 있지만 한 가지 문제가 존재한다.

  바로 '이벤트 발행은 성공하였지만, 이벤트 전달에는 실패'하는 경우가 발생할 수 있다는 것이다.

  이벤트가 발행된 다음, 이벤트 전달(부가 로직)에는 실패할 경우 해당 이벤트는 이미 소비된 이후 상태이기 때문에 부가 로직을 재시도할 수도 없게 된다.

  Kafka, RabbitMQ와 같은 메시지브로커에서는 실패한 메시지(Dead Letter)에 대한 재시도 로직을 수행할 수 있기는 하지만, 메시지브로커 자체에서 문제가 발생한다면 메시지브로커에 전달된 이벤트 조차도 유실될 수 있다는 문제점이 존재한다.

  따라서, 이벤트를 이벤트 버스로 발행하기에 앞서 이벤트를 데이터베이스에 기록하고 상태를 기록하여 실패한 이벤트를 재발행하여 최소한 한 번은 실행을 보장(ALO)할 수 있게 되는 것이다.

  이것이 이벤트 기반 아키텍처에서 트랙잭션 아웃박스 패턴이 언급되고 필요한 이유이다.

  다음 포스팅에서는 트랙잭션 아웃박스 패턴을 적용하여 비밀번호 초기화 메일 전송 로직이 전송 보장되도록 리팩토링하는 과정에 대한 내용을 작성할 예정이다.

  최근 모의 주식 투자 서비스 ‘무주시(무자본 주식 시뮬레이션)’를 되돌아보며 과거의 부족함을 느끼고, 구조 및 기능적으로 개선할 부분을 리팩토링하고 있다. 해당 리팩토링의 가장 큰 목표는 ‘한국투자증권 웹소켓 호출(구독 수) 유량 제한 정책 극복’에 있다. 한국투자증권에서는 기본적으로 API나 웹소켓에 대하여 호출 유량 제한 정책을 시행하고 있다. 웹소켓의 경우에는 하나의 세션 당 41개의 종목까지 구독이 가능하다. 또한, 하나의 세션은 하나의 개발자센터 계좌만 사용하기 때문에 실시간 체결가와 같은 정보를 제공하는데에는 어려움이 있다.

  이전부터 해당 문제에 대한 개선을 시도하였지만, 웹소켓 세션 관리 방법에 대한 지식 부족과 웹소켓 관련 시스템 구조가 만족스럽지 않아 해당 PR을 병합해보지는 못하였다. 이번 기회를 통하여 웹소켓 세션 관리 방법을 중점적으로 하여 리팩토링을 심도있게 진행해보기로 하였다. 프로젝트를 같이하는 팀원이 있기에 해당 팀원의 계좌를 추가적으로 사용하여 2개의 계좌로 웹소켓 접속키를 발급하여 2개의 웹소켓 세션을 운용하여 총 구독 가능 종목 수를 41 * 2 = 82 개로 늘려보고자 한다.

  웹소켓 세션과 구독 가능 종목 수를 늘리는 문제 외에도 기존 코드에서는 WebSocketSession을 저장하기 위해서 static 변수로 선언해야지만 웹소켓 세션을 저장할 수 있었다. 해당 동작에 의문을 느꼈지만 정확한 동작을 이해하지 못하였기에 단순히 “WebSocketSession은 static으로 선언하여야 직접 관리할 수 있구나”라고 넘겨짚었었다. 이번 리팩토링을 통해 웹소켓 세션 연결부터의 과정과 사용되는 클래스와 메서드들을 자세히 살펴보며 해당 문제도 해결할 수 있게 되었다. 이 과정을 공유해보고자 한다.

클라이언트 - 서버 - 한국투자증권 웹소켓 흐름

  클라이언트는 한국 주식 시장이 열렸을 때 특정 종목 상세 페이지에 접속하게 된다면 해당 종목에 대한 실시간 체결가를 확인할 수 있게 된다. 따라서, 클라이언트 - 서버 - 한국투자증권은 웹소켓으로 연결되어 있다. 또한, 한국투자증권 실시간 체결가 웹소켓에 클라이언트가 현재 확인하고 있는 주식 종목에 대한 구독을 요청하게 된다.

1. [클라이언트 → 서버] 구독 요청

   한국 주식 시장이 열려있을 때, 클라이언트가 특정 종목의 상세 페이지에 접속하면 해당 종목에 대한 구독 요청을 위하여 서버로 웹소켓 연결을 시도한다.

2. [서버] 구독 종목 추가

   만약, 특정 종목을 구독하고 있는 사용자가 존재하지 않는데 특정 종목에 대한 실시간 체결가를 계속해서 수신받게 된다면 리소스 낭비로 이어질 것이다. 따라서, 서버에서는 클라이언트들이 구독하고 있는 종목의 구독 수를 관리해 현재 구독하고 있는 종목에 대해서만 구독 요청을 보내야한다.

   따라서, 클라이언트의 특정 종목에 대한 구독 요청이 오면 서버에서는 해당 종목을 현재 관리 중인 구독 종목 목록에 추가해야한다.

3. [서버 → 한국투자증권] 국내주식 실시간 체결가 특정 종목 구독 요청

   서버에서는 특정 주식 종목이 처음 구독된 경우 한국투자증권 국내주식 실시간 체결가 웹소켓을 통해 해당 종목 구독 요청을 보낸다.

4. [한국투자증권 → 서버] 국내주식 실시간 체결가 응답

   한국투자증권에서는 구독한 종목에 대한 실시간 체결가를 웹소켓 세션을 통해 송신한다.

5. [서버 → 메시지브로커] 실시간 체결가 발행

   특정 종목을 구독 중인 클라이언트들에게 해당 종목의 실시간 체결가를 안정적으로 전달하기 위해 메시지브로커를 사용한다. 따라서, 한국투자증권으로부터 수신받은 실시간 체결가를 메시지 브로커로 발행한다.

6. [메시지브로커 → 서버] 실시간 체결가 수신

   서버에서는 메시지브로커에서 발행된 메시지를 수신한다.

   이 과정은 사실 메시지의 발행 모듈과 수신 모듈을 다른 모듈에 배치시켜야하지만 현재는 서버 비용 등의 문제로 인하여 동일한 모듈 내에 배치시켰다. 이것또한 차후 개선 사항이다.

7. [서버 → 클라이언트] 웹소켓 메시지 수신

   서버에서는 메시지브로커로부터 수신된 실시간 체결가 메시지를 받아 특정 주식 종목을 구독 중인 클라이언트에게 해당 종목의 실시간 체결가를 전달한다.

  클라이언트가 보고있는 특정 주식 종목에 대한 실시간 체결가를 관리하고, 안정적인 메시지 전달을 위해 위와 같은 프로세스를 구상하였다. 메시지브로커로의 메시지 발행 모듈과 수신 모듈이 같은 모듈 내에 있다는 점 등 구조적인 개선사항은 존재하지만 현재는 서버 비용 등의 문제로 인하여 위와 같은 구조로 설계하였다. 또한, 구독 중인 종목의 구독 해지 과정도 동일한 프로세스로 진행된다.

  위의 프로세스에서 가장 중요한 부분은 구독 종목 관리와 한국투자증권과 연결된 세션 당 현재 구독 수를 관리하는 것이다. 따라서, 아래와 같은 부분은 중점으로하여 코드를 구상하였다.

• 웹소켓 세션 : 웹소켓 접속키 = 1 : 1 로 대응되어 관리되어야 한다.
• 특정 종목이 이미 구독되고 있을 경우에는 한국투자증권에 추가적인 구독 요청이 필요없다.
• 특정 종목에 대한 구독 해제 후, 해당 종목에 대한 구독 수가 0이면 한국투자증권에 구독해제 요청을 보내야한다.
• 해당 종목을 구독하여 메시지를 수신받고 있는 웹소켓 세션을 알아야 이후 구독 해제 요청이 가능하다.

기존 코드 및 구조의 문제점

1. WebSocketConnectionManager를 사용한 웹소켓 세션 연결 관리

@Configuration
@EnableWebSocket
@RequiredArgsConstructor
public class KisWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    private final KisProperties kisProperties;
    private final KisRealTimeTradeHandler kisRealTimeTradeHandler;

    @Bean
    public WebSocketConnectionManager webSocketConnectionManager() {
        WebSocketClient client = new StandardWebSocketClient();
        return new WebSocketConnectionManager(
                client,
                kisRealTimeTradeHandler,
                kisProperties.getWebSocketDomain()
        );
    }
}

@Component
@DependsOn("kisWebSocketConfig")
@RequiredArgsConstructor
public class WebSocketConnectionScheduler {
    private final WebSocketConnectionManager connection;
    private final KisWebSocketHandler kisRealTimeTradeHandler;

    // ...

    @Scheduled(cron = "0 59 8 * * 1-5")
    public void runConnectWebSocketSessionJob() {
        connection.start();
    }

    @Scheduled(cron = "0 30 15 * * 1-5")
    public void runDisconnectWebSocketToKis() {
        if (connection.isConnected())
            connection.stop();
    }
    
    // ...
}

  기존 코드에서는 위와 같이 WebSocketConnectionManager를 빈으로 등록하여서 웹소켓 연결을 관리하였다.

  WebSocketConnectionManager를 사용하여 웹소켓 연결을 관리한 이유는 해당 작업을 할 때 여러 블로그들에서 WebSocketConnectionManager를 사용하여 웹소켓 세션을 연결하는 코드를 보았고, TextWebSocketHandler.afterConnectionEstablished(WebSocketSession) 메서드에서 웹소켓 세션을 인자로 받고 있기 때문에 필자는 “TextWebSocketHandler를 통해서 웹소켓 세션을 받아와야하는구나!”라는 생각을 했었기 때문이다.

  그러나, 이는 웹소켓 세션의 연결 과정을 잘 이해를 하지 못하였기 때문에 발생한 이슈이다.

  WebSocketConnectionManager는 웹소켓 연결과 세션 관리를 별도로 할 필요없도록 관련 기능들을 추상화한 클래스이다.

public class WebSocketConnectionManager extends ConnectionManagerSupport {
    private final WebSocketClient client;
    private final WebSocketHandler webSocketHandler;
    @Nullable
    private WebSocketSession webSocketSession;

    // ...
}

  WebSocketConnectionManger의 내부를 확인해보면 WebSocketSession을 멤버 변수로 가지고 있다는 것을 알 수 있다. 즉, 단순히 메서드를 호출하여 웹소켓 세션을 연결/종료할 수 있다는 장점은 있지만, 객체 내부에서 웹소켓 세션을 관리하여 클라이언트가 직접 관리하는 등의 작업을 하기에는 적합하지 않다.

  웹소켓 연결 요청을 진행하는 실질적인 클래스(인터페이스)는 WebSocketClient이다.

  WebSocketConnectionManager 객체를 생성할 때도 인자로 WebSocketClient의 구현체를 넘겨주듯이 웹소켓 연결/종료를 직접적으로 수행하고 연결된 웹소켓 세션을 반환하는 클래스이다.

  WebSocketClient는 내부에 .dohandshake()와 .execute() 메서드를 가지고 있으며, 각 반환값은 Future<WebSocketSession>이다.

  해당 메서드들이 실제로 웹소켓을 연결하고, 세션을 반환하는 기능을 담당하고 있다. 따라서, WebSocketClient를 통해 직접 웹소켓을 연결하여야지만 제대로 웹소켓 세션을 반환받아 관리할 수 있는 것이다.

  기존 코드에서는 TextWebSocketHandler에서 메서드의 인자로 넘겨오는 웹소켓 세션을 관리하였지만, 이것은 클래스의 책임에도 맞지 않는 구조이며 제대로된 관리 방법이라 할 수 없다.

  또한, 각 메서드의 반환값이 Future<WebSocketSession> 임을 알 수 있는데, 이는 아래의 2. WebSocketSession을 static 멤버 변수로 선언해야지만 웹소켓 세션이 바인딩 가능하던 문제의 원인이기도 하다.

2. WebSocketSession을 static 멤버 변수로 선언해야지만 웹소켓 세션이 바인딩 가능하던 문제

@Slf4j
public abstract class KisWebSocketHandler extends TextWebSocketHandler {
    protected static WebSocketSession session;
    protected final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 웹 소켓 연결 지속을 위한 메서드
     *
     * - 연결 유지를 위한 PINGPONG 메시지 송신
     */
    public void keepConnection() {
        Map<String, String> header = new HashMap<>();
        header.put("tr_id", "PINGPONG");
        header.put("datetime", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss")));
        Map<String, Object> input = new HashMap<>();
        input.put("header", header);
        try {
            session.sendMessage(new TextMessage(objectMapper.writeValueAsString(input)));
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    public abstract void connect(String stockCode);
    public abstract void disconnect(String stockCode);
}

@Component
@Slf4j
@RequiredArgsConstructor
public class KisRealTimeTradeHandler extends KisWebSocketHandler {
    private final TradeNotificationPublisher tradeNotificationPublisher;
    private final ObjectMapper objectMapper;
    private final RedisService redisService;

    private static final String TR_ID = "H0STCNT0";
    private static final int MAX_CONNECTION = 41;
    private final ConcurrentHashMap<String, Integer> subscribedStocks = new ConcurrentHashMap<>(MAX_CONNECTION);

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        this.session = session;
    }

    // ...
}

  앞서 확인했듯이 위의 기존 코드는 TextWebSocketHandler의 구현체에서 웹소켓 세션을 직접 관리하였기에 클래스의 책임도 너무 크며, 좋지 못한 구조였다.

  그러나, 해당 상황에서 발생한 ‘WebSocketSession을 static 변수로 선언하여지만 값이 바인딩되던 문제’에 대해 알아보고자 한다.

  해당 문제의 원인은 위에서 확인하였던 WebSocketClient.execute()의 반환값이 CompletableFuture<WebSocketSession>이기 때문이다. 따라서, static 변수로 선언하여야지만 이후 비동기적으로 연결이 완료된 후 반환되는 세션을 받을 수 있었던 것이다.

  필자는 이번 리팩토링에서 WebSocketClient.execute()를 직접 사용하여 웹소켓 세션을 연결하고 관리할 것이기 때문에 CompletableFuture<WebSocketSession> 반환값에서 WebSocketSession 객체를 얻는 방법을 알아보았다.

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
    // ...
    /**
     * Returns the result value when complete, or throws an
     * (unchecked) exception if completed exceptionally. To better
     * conform with the use of common functional forms, if a
     * computation involved in the completion of this
     * CompletableFuture threw an exception, this method throws an
     * (unchecked) {@link CompletionException} with the underlying
     * exception as its cause.
     *
     * @return the result value
     * @throws CancellationException if the computation was cancelled
     * @throws CompletionException if this future completed
     * exceptionally or a completion computation threw an exception
     */
    @SuppressWarnings("unchecked")
    public T join() {
        Object r;
        if ((r = result) == null)
            r = waitingGet(false);
        return (T) reportJoin(r);
    }
    // ...
}

  CompletableFuture<T> 클래스에서는 get(long timeout, TimeUnit unit) 메서드를 사용해 시간 내에 값을 받아오는 방법도 존재하고, join() 메서드를 통해 비동기 작업이 완료된 후 값을 받아오는 방법이 존재한다. 필자는 웹소켓 세션 연결이 필수적으로 되어야지만 실시간 체결가 제공이 가능하기 때문에 .join() 메서드를 사용하여 웹소켓 세션을 받아오기로 하였다.

본론

관련 PR

Refact: 한국투자증권 웹소켓 연결 세션 증설 및 구독 관리 로직 리팩토링

  서론에서 이야기하였던 구조를 바탕으로 아래와 같은 기능(책임)을 담당하는 클래스들을 정의하였다.

• StompInterceptor: 클라이언트에서 서버로 보낸 웹소켓 메시지를 처리하는 클래스
• TradeNotificationPublisher: Reids pub/sub 메시지 브로커 방식을 통해 메시지를 발행하는 클래스
• KisRealTimeTradeWebSocketHandler: 한국투자증권으로부터 웹소켓 세션을 통해 수신되는 메시지를 처리하는 클래스
• KisWebSocketConnector: 한국투자증권 웹소켓 세션 연결을 담당하는 클래스
• KisWebSocketSessionManager: 한국투자증권과 연결된 웹소켓 세션을 관리하는 클래스
• KisSubscriptionManager: 주식 구독 종목 및 종목 당 연결 세션 정보 관리 클래스
• KisRealTimeTradeWebSocketClient: 한국투자증권 국내주식 실시간 체결가 웹소켓 요청 클래스

  위 클래스들 중 StompInterceptor와 TradeNotificationPublisher는 기존에 존재하던 클래스로, 이번 작업들 통해 약간의 코드 변경이 존재하였다.

  그 외는 이번 작업을 통해서 코드를 완전히 변경하거나, 새로 만든 클래스들이다.

  구독 로직의 전체적인 로직은 다음과 같다.

  초기에 한국투자증권과 웹소켓 세션을 연결하는 KisWebSocketConnector는 해당 시퀀스 다이어그램에서 제외되었다. KisWebSocketConnector의 로직에 대해서는 아래에서 자세히 설명한다.

1. KisWebSocketConnector

@Slf4j
@Component
public class KisWebSocketConnector {
    private final WebSocketClient webSocketClient = new StandardWebSocketClient();
    private final KisRealTimeTradeWebSocketHandler kisRealTimeTradeWebSocketHandler;
    private final String webSocketDomain;
    
    public KisWebSocketConnector(
            KisProperties kisProperties,
            KisRealTimeTradeWebSocketHandler kisRealTimeTradeWebSocketHandler
    ) {
        this.webSocketDomain = kisProperties.getWebSocketDomain();
        this.kisRealTimeTradeWebSocketHandler = kisRealTimeTradeWebSocketHandler;
    }
    
    /**
     * 한국투자증권 웹소켓 세션 연결 메서드
     *
     * @return  한국투자증권 웹소켓과 연결된 세션
     */
    public WebSocketSession connect() {
        try {
            WebSocketSession session = webSocketClient
                    .execute(kisRealTimeTradeWebSocketHandler, webSocketDomain).join();
            
            return session;
        } catch (Exception e) {
            log.error("[Error] Failed to connect to KIS WebSocket - {}", e.getMessage());
            return null;
        }
    }
}

  KisWebSocketConnector는 한국투자증권 웹소켓 도메인을 통해 웹소켓 세션을 연결한 후, 해당 웹소켓 세션을 반환하는 클래스이다.

  앞서, 서론의 2. WebSocketSession을 static 멤버 변수로 선언해야지만 웹소켓 세션이 바인딩 가능하던 문제에서 설명하였듯이, 웹소켓을 연결하고 반환된 세션을 직접 관리하기 위해 WebSocketClient.execute() 메서드를 사용하여 연결 후 반환된 세션을 직접 받아 반환한다.

  해당 웹소켓 세션 연결 메서드는 주식 시장 시작 시에 KisWebSocketSessionManager의 세션 초기화 시 호출된다. 반환된 세션은 KisWebSocketSessionManager에서 관리하게 된다.

2. KisRealTimeTradeWebSocketHandler

@Slf4j
@Component
@RequiredArgsConstructor
public class KisRealTimeTradeWebSocketHandler extends TextWebSocketHandler {
    private final TradeNotificationPublisher tradeNotificationPublisher;
    
    /**
     * 웹소켓 세션 연결 후 실행 메서드
     *
     * @param session       한국투자증권 웹소켓과 연결된 세션
     * @throws Exception    웹소켓 세션 연결 시 발생 예외
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("KIS Websocket session connected: {}", session.getId());
        super.afterConnectionEstablished(session);
    }
    
    /**
     * 웹소켓 세션 종료 후 실행 메서드
     *
     * @param session       한국투자증권 웹소켓과 연결되었던 세션
     * @param status        웹소켓 세션 연결 종료 상태
     * @throws Exception    웹소켓 세션 연결 종료 시 발생 예외
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("KIS Websocket session closed: {}", session.getId());
        super.afterConnectionClosed(session, status);
    }
    
    /**
     * 한국투자증권 웹소켓 세션을 통해 전달받은 메시지를 처리하는 메서드
     *
     * <p> 구독 주식 종목 실시간 체결가 메시지 수신 시, 해당 종목 구독자에게 메시지 송신
     *
     * <p> 세션 연결 유지를 위한 핑퐁(PingPong) 메시지 수신 시, 수신받은 페이로드와 동일한 페이로드를 응답
     *
     * @param session       웹소켓 세션
     * @param message       수신 메시지
     * @throws Exception    메시지 수신 시 발생 예외
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        
        if (isPingPong(payload)) {
            session.sendMessage(new TextMessage(payload));
            return;
        }
        
        if (isMetaMessage(payload)) {
            if (isErrorMessage(payload)) {
                log.error("[Error] Error message from KIS websocket - {}", payload);
            } else {
                log.info("[KIS Websocket] - {}", payload);
            }
            return;
        }
        
        tradeNotificationPublisher.publishTradeNotification(parsePayloadToNotificationDto(payload));
    }
    
    /**
     * 페이로드의 핑퐁 메시지 여부를 확인하는 메서드
     *
     * @param payload   페이로드
     * @return          페이로드의 핑퐁 메시지 여부
     */
    private boolean isPingPong(String payload) {
        return payload.contains("PINGPONG");
    }
    
    /**
     * 페이로드의 메타 메시지 여부를 확인하는 메서드
     *
     * @param payload   페이로드
     * @return          페이로드의 메타 메시지 여부
     */
    private boolean isMetaMessage(String payload) {
        return payload != null && !payload.isBlank() && payload.startsWith("{");
    }
    
    /**
     * 페이로드의 에러 메시지 여부를 확인하는 메서드
     *
     * <p> 웹소켓 응답 메타 메시지의 반환 코드(rt_cd)가 0을 제외한 나머지 경우는 모두 에러 응답
     *
     * @param payload   페이로드
     * @return          페이로드의 에러 메시지 여부
     */
    private boolean isErrorMessage(String payload) {
        return !payload.contains("\"rt_cd\":\"0\"");
    }
    
    /**
     * 페이로드를 해당 주식 종목 구독자에게 전달하기 위한 객체({@link TradeNotificationDto})로 변환하는 메서드
     *
     * @param payload   페이로드
     * @return          {@link TradeNotificationDto} 객체 리스트
     */
    private List<TradeNotificationDto> parsePayloadToNotificationDto(String payload) {
        List<TradeNotificationDto> result = new ArrayList<>();
        String[] parts = payload.split("\\^");
        String[] metas = parts[0].split("\\|");
        int tradeCount = Integer.parseInt(metas[2]);
        String stockCode = metas[3];
        
        for (int idx = 0, i = 0; i < tradeCount; i++) {
            result.add(TradeNotificationDto.builder()
                    .stockCode(stockCode)
                    .time(convertTime(parts[idx + 1]))
                    .price(Long.valueOf(parts[idx + 2]))
                    .stockCount(Long.valueOf(parts[idx + 12]))
                    .volume(Long.valueOf(parts[idx + 13]))
                    .tradeType((parts[idx + 21].equals("1")) ? TradeType.BUY : TradeType.SELL)
                    .changeRate(Double.valueOf(parts[idx + 5]))
                    .build());
            idx += 46;
        }
        
        return result;
    }
    
    /**
     * 페이로드에서 전달된 시각을 양식에 맞게 변환하는 메서드
     *
     * @param time  페이로드에 전달된 시각
     * @return      양식에 맞게 변환된 시각
     */
    private String convertTime(String time) {
        return time.substring(0, 2) + ":" + time.substring(2, 4) + ":" + time.substring(4);
    }
}

  KisRealTimeTradeWebSocketHandler는 웹소켓 세션을 통해 수신되는 한국투자증권 국내주식 실시간 체결가 메시지를 처리하는 클래스이다.

  handleTextMessage(WebSocketSession session, TextMessage message) 메서드를 통해 수신되는 메시지의 페이로드를 분석하여 후처리를 진행한다.

  수신되는 페이로드가 핑퐁 메시지인 경우 한국투자증권 개발자센터 공지에 따라, 동일한 페이로드를 응답한다. 또한, 에러 메시지의 경우네느 에러 메시지 로그를 출력하도록 한다.

  해당 메시지가 유효한 메시지(실시간 체결가)인 경우에는 해당 메시지를 DTO에 맞도록 파싱하여 TradeNotificationPublisher를 통해 메시지브로커로 해당 메시지를 발행한다.

3. TradeNotificationPublisher

@Component
@RequiredArgsConstructor
public class TradeNotificationPublisher {
    private final RedisTemplate redisTemplate;
    
    /**
     * 한국투자증권 국내 주식 체결가 정보 목록을 Redis Pub/Sub 토픽으로 발행(Publish)하는 메서드
     *
     * @param tradeNotifications    주식 체결가 정보 목록
     */
    public void publishTradeNotification(List<TradeNotificationDto> tradeNotifications) {
        for (TradeNotificationDto tradeNotification : tradeNotifications) {
            this.publishTradeNotification(tradeNotification);
        }
    }
    
    /**
     * 한국투자증권 국내 주식 체결가 정보를 Redis Pub/Sub 토픽으로 발행(Publish)하는 메서드
     *
     * @param tradeNotification     주식 체결가 정보
     */
    private void publishTradeNotification(TradeNotificationDto tradeNotification) {
        redisTemplate.convertAndSend(ChannelConstant.TRADE.getValue(), tradeNotification);
    }
}

  TradeNotificationPublisher는 Redis pub/sub 메시지 브로커를 사용한 주식 체결가 정보를 발행하는 역할을 담당하는 클래스이다.

4. KisWebSocketSessionManager

@Slf4j
@Component
@RequiredArgsConstructor
public class KisWebSocketSessionManager {
    private final Map<String, KisWebSocketSession> sessions = new HashMap<>();
    private final KisWebSocketConnector kisWebSocketConnector;
    private final KisAuthService kisAuthService;
    
    /**
     * 웹소켓 세션을 초기화하는 메서드
     *
     * <p> {@link KisWebSocketConnector}를 통하여 한국투자증권 웹소켓과 연결 후 반환된 세션을 저장
     *
     * @return 한국투자증권 웹소켓과 연결된 세션의 아이디 목록
     */
    public List<String> initializeSessions() {
        List<String> webSocketKeys = kisAuthService.getWebSocketKeys();
        
        for (String webSocketKey : webSocketKeys) {
            WebSocketSession webSocketSession = kisWebSocketConnector.connect();
            
            if (webSocketSession != null) {
                sessions.put(webSocketSession.getId(), new KisWebSocketSession(webSocketSession, webSocketKey));
            }
        }
        
        return sessions.keySet().stream().toList();
    }
    
    /**
     * 웹소켓 세션 종료 후, 삭제하는 메서드
     */
    public void closeSessions() {
        for (KisWebSocketSession kisWebSocketSession : sessions.values()) {
            try {
                kisWebSocketSession.getWebSocketSession().close();
            } catch (IOException e) {
                log.error("[Error] Failed to close Websocket session - {}", kisWebSocketSession.getWebSocketSession().getId());
            }
        }
        
        sessions.clear();
    }
    
    /**
     * 특정 한국투자증권 웹소켓 세션 정보 객체를 조회하는 메서드
     *
     * @param sessionId 조회할 웹소켓 세션 ID
     * @return          한국투자증권 웹소켓 세션 정보 객체
     */
    public KisWebSocketSession getKisWebSocketSession(String sessionId) {
        return sessions.get(sessionId);
    }
    
    /**
     * 한국투자증권과 연결된 웹소켓 세션과 해당 세션의 웹소켓 접속키를 저장하는 클래스
     */
    public static class KisWebSocketSession {
        private final WebSocketSession webSocketSession;
        private final String webSocketKey;
        
        public KisWebSocketSession(WebSocketSession webSocketSession, String webSocketKey) {
            this.webSocketSession = webSocketSession;
            this.webSocketKey = webSocketKey;
        }
        
        public WebSocketSession getWebSocketSession() {
            return this.webSocketSession;
        }
        
        public String getWebSocketKey() {
            return this.webSocketKey;
        }
    }
}

  KisWebSocketSessionManager는 한국투자증권과 연결된 웹소켓 세션을 관리하는 클래스이다.

  .initializeSessions() 메서드에서는 웹소켓 세션을 저장하는 초기화 메서드이다. KisAuthService.getWebSocketKeys()를 호출하여 현재 저장된 모든 한국투자증권 웹소켓 접속키만큼 웹소켓 세션을 연결해 저장한다. 멤버 변수인 kisWebSocketConnector.connect() 메서드를 호출하여 반환된 웹소켓 세션을 저장한다.

  따라서, 계좌(웹소켓 접속키)가 여러 개인 경우더라도 해당 웹소켓 접속키만큼의 웹소켓 세션이 연결 가능하다.

  한국투자증권 웹소켓 세션은 세션 당 하나의 계좌(웹소켓 접속키)만 사용이 가능하기 때문에, 웹소켓 접속키:웹소켓 세션 = 1 : 1로 관리되어야한다. 따라서, 내부 클래스인 KisWebSocketSession은 웹소켓 세션과 웹소켓 접속키를 함께 저장한다.

5. KisSubscriptionManager

@Component
@RequiredArgsConstructor
public class KisSubscriptionManager {
    private final KisWebSocketSessionManager kisWebSocketSessionManager;
    private final KisRealTimeTradeWebSocketClient kisRealTimeTradeWebSocketClient;
    
    /**
     * 웹 소켓 세션 ID를 Key로 사용하여, 세션 별 구독 종목 정보({@link StockSubscriptionContext})를 저장하는 Map
     */
    private final Map<String, StockSubscriptionContext> stockSubscriptionContextBySession = new LinkedHashMap<>();
    
    /**
     * 종목 코드(StockCode)를 Key로 사용하여, 특정 종목 코드의 구독을 담당하는 웹 소켓 세션 ID를 저장하는 Map
     *
     * <p>해당 주식 종목 코드를 구독하고 있는 웹소켓 세션 ID를 바로 알아내기 위한 역인덱싱 목적
     */
    private final Map<String, String> stockSessionIndex = new HashMap<>();
    
    private final ReentrantLock lock = new ReentrantLock();
    
    /**
     * 웹소켓 세션 ID 목록을 인자로 받아, 웹소켓 세션 ID 별 구독 목록 Map을 초기화하는 메서드
     *
     * @param sessionIds    초기화할 웹소켓 세션 ID 목록
     */
    public void initialize(List<String> sessionIds) {
        for (String sessionId : sessionIds) {
            stockSubscriptionContextBySession.put(sessionId, new StockSubscriptionContext());
        }
    }
    
    /**
     * 웹소켓 세션 ID 별 구독 목록 Map과 주식 종목 코드 별 할당된 웹소켓 세션 ID Map을 비우는 메서드
     */
    public void clearSubscriptions() {
        stockSubscriptionContextBySession.clear();
        stockSessionIndex.clear();
    }
    
    /**
     * 주식 종목을 구독하는 메서드
     *
     * <p> 주식 종목 구독 정보를 관리하고, 초기 구독 요청이 온 종목의 경우에는
     * {@link muzusi.infrastructure.kis.websocket.KisWebSocketSessionManager}에 요청을 위임하여 구독 요청
     *
     * @param stockCode 주식 종목 코드
     */
    public void subscribe(String stockCode) {
        lock.lock();
        
        try {
            String sessionId = stockSessionIndex.get(stockCode);
            
            if (sessionId == null) { // 해당 주식 종목을 처음 구독하는 경우, 한국투자증권 국내주식 실시간 체결가 구독 요청
                sessionId = getAvailableSessionId();
                
                KisWebSocketSessionManager.KisWebSocketSession kisWebSocketSession = kisWebSocketSessionManager.getKisWebSocketSession(sessionId);
                
                kisRealTimeTradeWebSocketClient.subscribe(
                        kisWebSocketSession.getWebSocketSession(),
                        kisWebSocketSession.getWebSocketKey(),
                        stockCode
                );
                stockSessionIndex.put(stockCode, sessionId);
            }
            
            StockSubscriptionContext context = stockSubscriptionContextBySession.get(sessionId);
            context.add(stockCode);
        } finally {
            lock.unlock();
        }
    }
    
    /**
     * 구독 가능한 웹소켓 세션의 아이디를 반환하는 메서드
     *
     * @return 사용 가능한 웹소켓 세션 ID
     * @throws CustomException StockErrorType.MAX_REQUEST_WEB_SOCKET - 더 이상 구독 가능한 세션이 없는 경우
     */
    private String getAvailableSessionId() {
        return stockSubscriptionContextBySession.entrySet().stream()
                .filter(entry -> entry.getValue().isAvailable())
                .map(entry -> entry.getKey())
                .findFirst()
                .orElseThrow(() -> new CustomException(StockErrorType.MAX_REQUEST_WEB_SOCKET));
    }
    
    /**
     * 구독 중인 주식 종목에 대한 구독을 해제하는 메서드
     *
     * <p> 주식 종목 구독 해제를 담당하고, 구독 해제 후 해당 종목에 대한 더 이상 구독이 없는 경우에는
     * {@link muzusi.infrastructure.kis.websocket.KisWebSocketSessionManager}에 요청을 위임하여 구독 해제 요청
     *
     * @param stockCode 주식 종목 코드
     */
    public void unsubscribe(String stockCode) {
        lock.lock();
        
        try {
            String sessionId = stockSessionIndex.get(stockCode);
            
            if (sessionId == null || sessionId.isBlank()) {
                throw new CustomException(StockErrorType.NOT_SUBSCRIBED_STOCK);
            }
            
            StockSubscriptionContext context = stockSubscriptionContextBySession.get(sessionId);
            int subscriptionCount = context.getSubscriptionCount(stockCode);
            
            if (subscriptionCount == 1) { // 구독 해제 이후 더 이상 구독 수가 없다면, 해당 구독 종목을 삭제
                KisWebSocketSessionManager.KisWebSocketSession kisWebSocketSession = kisWebSocketSessionManager.getKisWebSocketSession(sessionId);
                kisRealTimeTradeWebSocketClient.unsubscribe(
                        kisWebSocketSession.getWebSocketSession(),
                        kisWebSocketSession.getWebSocketKey(),
                        stockCode
                );
                
                stockSessionIndex.remove(stockCode);
            }
            
            context.remove(stockCode);
        } finally {
            lock.unlock();
        }
    }
    
    /**
     * 주식 종목 코드와 해당 주식 종목에 대한 구독 수를 저장하는 컨텍스트 클래스
     *
     * <p>한국투자증권 웹소켓 호출 유량 제한을 충족하는 만큼의 구독 수를 관리
     */
    public static class StockSubscriptionContext {
        private static final int MAX_SUBSCRIPTION = 41;
        private final Map<String, Integer> subscribedStocks = new HashMap<>(MAX_SUBSCRIPTION);
        
        /**
         * 특정 주식 종목 구독 메서드
         *
         * <p> 처음 구독한 경우, 해당 종목에 대한 구독 수는 1로 설정
         *
         * @param stockCode 구독할 주식 종목 코드
         * @throws CustomException StockErrorType.MAX_REQUEST_WEB_SOCKET - 해당 세션을 통해 더 이상 새로운 종목을 구독할 수가 없는 경우
         */
        public int add(String stockCode) {
            return subscribedStocks.compute(
                    stockCode,
                    (stock, subscriptionCount) -> {
                        if (subscriptionCount == null) {
                            if (!isAvailable()) {
                                throw new CustomException(StockErrorType.MAX_REQUEST_WEB_SOCKET);
                            }
                            return 1;
                        }
                        
                        return subscriptionCount + 1;
                    }
            );
        }
        
        /**
         * 특정 주식 종목 구독 해제 메서드
         *
         * @param stockCode 구독 해제할 주식 종목 코드
         * @return          구독 해제 후 해당 주식 종목 구독 수
         */
        public int remove(String stockCode) {
            Integer afterSubscriptionCount = subscribedStocks.compute(
                    stockCode,
                    (stock, subscriptionCount) -> {
                        if (subscriptionCount == null) {
                            return null;
                        }
                        
                        if (subscriptionCount == 1) {
                            return null;
                        }
                        
                        return subscriptionCount - 1;
                    }
            );
            
            return afterSubscriptionCount == null ? 0 : afterSubscriptionCount;
        }
        
        /**
         * 특정 주식 종목 구독 수 반환 메서드
         *
         * @param stockCode 주식 종목 코드
         * @return          해당 주식 종목 구독 수
         */
        public int getSubscriptionCount(String stockCode) {
            return subscribedStocks.get(stockCode);
        }
        
        /**
         * 현재 겍체가 더 이상 구독이 가능한지 여부를 반환하는 메서드
         *
         * @return 최대 구독 가능 수보다 작을 경우 true, 최대 구독 수보다 같거나 클 경우 false
         */
        public boolean isAvailable() {
            return this.subscribedStocks.size() < MAX_SUBSCRIPTION;
        }
    }
}

  KisSubscriptionManager는 한국투자증권 실시간 체결가 웹소켓 구독 종목 관리를 위한 핵심 클래스이다.

  ReentrantLock을 통한 동시성 제어를 기반으로 동작하며, 주식 종목에 대한 구독/해제 로직을 담당하고 있다. 또한, 멤버 변수인 KisWebSocketSessionManager에서 관리하는 세션을 받아와, 또다른 멤버 변수인 KisRealTimeTradeWebSocketClient를 통해 해당 세션으로 구독/해제 요청을 보낸다.

  또한, 다음과 같은 Map 멤버 변수들을 가지고 있다.

/**
 * 웹 소켓 세션 ID를 Key로 사용하여, 세션 별 구독 종목 정보({@link StockSubscriptionContext})를 저장하는 Map
 */
private final Map<String, StockSubscriptionContext> stockSubscriptionContextBySession = new LinkedHashMap<>();

/**
 * 종목 코드(StockCode)를 Key로 사용하여, 특정 종목 코드의 구독을 담당하는 웹 소켓 세션 ID를 저장하는 Map
 *
 * <p>해당 주식 종목 코드를 구독하고 있는 웹소켓 세션 ID를 바로 알아내기 위한 역인덱싱 목적
 */
private final Map<String, String> stockSessionIndex = new HashMap<>();

  stockSubscriptionContextSession은 Key로 웹소켓 세션 ID, Value로 StockSubscriptionContext를 가지고 있다.

  StockSubscriptionContext는 구독 종목과 해당 구독 종목의 구독 수를 관리하는 클래스이다. stockSubscriptionContextSession Map으로 저장되기 때문에 즉, 세션 당 구독 종목을 관리하게 되는 것이다.

  앞서 설명하였듯이, 특정 주식 종목이 이미 구독 되고 있는 경우 해당 종목은 추가적인 구독 요청을 진행하지 않아도 된다. 따라서, Key로 주식 종목 코드, Value로 해당 종목이 구독되고 있는 웹소켓 세션 ID를 사용하는 역인덱싱 목적의 stockSessionIndex Map을 가지고 있다.

/**
 * 주식 종목을 구독하는 메서드
 *
 * <p> 주식 종목 구독 정보를 관리하고, 초기 구독 요청이 온 종목의 경우에는
 * {@link muzusi.infrastructure.kis.websocket.KisWebSocketSessionManager}에 요청을 위임하여 구독 요청
 *
 * @param stockCode 주식 종목 코드
 */
public void subscribe(String stockCode) {
    lock.lock();
    
    try {
        String sessionId = stockSessionIndex.get(stockCode);
        
        if (sessionId == null) { // 해당 주식 종목을 처음 구독하는 경우, 한국투자증권 국내주식 실시간 체결가 구독 요청
            sessionId = getAvailableSessionId();
            
            KisWebSocketSessionManager.KisWebSocketSession kisWebSocketSession = kisWebSocketSessionManager.getKisWebSocketSession(sessionId);
            
            kisRealTimeTradeWebSocketClient.subscribe(
                    kisWebSocketSession.getWebSocketSession(),
                    kisWebSocketSession.getWebSocketKey(),
                    stockCode
            );
            stockSessionIndex.put(stockCode, sessionId);
        }
        
        StockSubscriptionContext context = stockSubscriptionContextBySession.get(sessionId);
        context.add(stockCode);
    } finally {
        lock.unlock();
    }
}

/**
 * 구독 가능한 웹소켓 세션의 아이디를 반환하는 메서드
 *
 * @return 사용 가능한 웹소켓 세션 ID
 * @throws CustomException StockErrorType.MAX_REQUEST_WEB_SOCKET - 더 이상 구독 가능한 세션이 없는 경우
 */
private String getAvailableSessionId() {
    return stockSubscriptionContextBySession.entrySet().stream()
            .filter(entry -> entry.getValue().isAvailable())
            .map(entry -> entry.getKey())
            .findFirst()
            .orElseThrow(() -> new CustomException(StockErrorType.MAX_REQUEST_WEB_SOCKET));
}

  특정 주식 종목 구독과 해제 로직은 모두 ReentrantLock을 통해 동시성을 제어한다.

  락 획득 후 우선, 역인덱스 stockSessionIndex를 통해 해당 종목이 이미 구독 중인 웹소켓 세션의 ID를 획득한다. 이 때, 웹소켓 세션 ID가 null인 경우에는 해당 주식 종목을 처음 구독하는 경우이기 때문에 한국투자증권 국내주식 실시간 체결가 웹소켓에 구독 요청을 보낸다.

  구독이 가능한 웹소켓 세션 ID를 가져온 다음, 해당 세션 ID를 통해 kisWebSocketSessionManager으로부터 해당 세션을 획득한다. 이후, 해당 세션과 kisRealTimeTradeWebSocketClient를 통해 한국투자증권 실시간 체결가 구독 요청을 보낸다. 이후, 역인덱스 stockSessionIndex에 해당 주식 종목 코드와 해당 종목이 구독되어 메시지를 전달받고 있는 세션 ID를 저장한다.

  최종적으로, 최초 구독 요청 여부와 상관없이 stockSubscriptionContextBySession Map에서 해당 세션에 대한 특정 종목 구독 수를 증가시킨다.

// KisSubscriptionManager.StockSubscriptionContext
/**
 * 특정 주식 종목 구독 메서드
 *
 * <p> 처음 구독한 경우, 해당 종목에 대한 구독 수는 1로 설정
 *
 * @param stockCode 구독할 주식 종목 코드
 * @throws CustomException StockErrorType.MAX_REQUEST_WEB_SOCKET - 해당 세션을 통해 더 이상 새로운 종목을 구독할 수가 없는 경우
 */
public int add(String stockCode) {
    return subscribedStocks.compute(
            stockCode,
            (stock, subscriptionCount) -> {
                if (subscriptionCount == null) {
                    if (!isAvailable()) {
                        throw new CustomException(StockErrorType.MAX_REQUEST_WEB_SOCKET);
                    }
                    return 1;
                }
                
                return subscriptionCount + 1;
            }
    );
}

/**
 * 특정 주식 종목 구독 해제 메서드
 *
 * @param stockCode 구독 해제할 주식 종목 코드
 * @return          구독 해제 후 해당 주식 종목 구독 수
 */
public int remove(String stockCode) {
    Integer afterSubscriptionCount = subscribedStocks.compute(
            stockCode,
            (stock, subscriptionCount) -> {
                if (subscriptionCount == null) {
                    return null;
                }
                
                if (subscriptionCount == 1) {
                    return null;
                }
                
                return subscriptionCount - 1;
            }
    );
    
    return afterSubscriptionCount == null ? 0 : afterSubscriptionCount;
}

  KisSubscriptionManager의 내부 클래스인 StockSubscriptionContext에서는 구독 종목 추가, 삭제 시 자체적인 처리 로직을 가진다. 따라서, 웹소켓 호출 유량 제한 내에서 유동적으로 구독 종목을 관리할 수 있게 된다.

  해당 절에서 따로 설명하지는 않았지만, 구독 해제 로직의 경우에도 거의 동일한 흐름으로 로직이 진행된다.

6. KisRealTimeTradleWebSocketClient

@Slf4j
@Component
@RequiredArgsConstructor
public class KisRealTimeTradeWebSocketClient {
    private final ObjectMapper objectMapper;
    private static final String TRADE_ID = "H0STCNT0";
    
    private enum TradeType {
        SUBSCRIPTION(1),
        UNSUBSCRIPTION(2);
        
        private final int value;
        
        TradeType(int value) {
            this.value = value;
        }
        
        public int getValue() {
            return this.value;
        }
    }
    
    /**
     * 한국투자증권 국내주식 실시간 체결가 웹소켓 구독 요청 메서드
     *
     * @param session       웹소켓 세션
     * @param webSocketKey  웹소켓 접속키
     * @param stockCode     주식 종목 코드
     */
    public void subscribe(WebSocketSession session, String webSocketKey, String stockCode) {
        this.request(session, webSocketKey, stockCode, TradeType.SUBSCRIPTION);
    }
    
    /**
     * 한국투자증권 국내주식 실시간 체결가 웹소켓 구독 해제 요청 메서드
     *
     * @param session       웹소켓 세션
     * @param webSocketKey  웹소켓 접속키
     * @param stockCode     주식 종목 코드
     */
    public void unsubscribe(WebSocketSession session, String webSocketKey, String stockCode) {
        this.request(session, webSocketKey, stockCode, TradeType.UNSUBSCRIPTION);
    }
    
    /**
     * 한국투자증권 국내주식 실시간 체결가 웹소켓 요청 메서드
     *
     * @param session       웹소켓 세션
     * @param webSocketKey  웹소켓 접속키
     * @param stockCode     주식 종목 코드
     * @param tradeType     거래 타입 (1: 구독, 2: 해제)
     */
    private void request(WebSocketSession session, String webSocketKey, String stockCode, TradeType tradeType) {
        if (session == null || !session.isOpen()) {
            log.error("[Error] Failed to send request KIS Websocket - Session is null or closed.");
            return;
        }
        
        Map<String, String> header = new HashMap<>();
        header.put("approval_key", webSocketKey);
        header.put("custtype", "P");
        header.put("tr_type", String.valueOf(tradeType.getValue()));
        header.put("content-type", "utf-8");
        
        Map<String, Object> body = new HashMap<>();
        Map<String, String> input = new HashMap<>();
        input.put("tr_id", TRADE_ID);
        input.put("tr_key", stockCode);
        body.put("input", input);
        
        Map<String, Object> request = new HashMap<>();
        request.put("header", header);
        request.put("body", body);
        
        try {
            session.sendMessage(new TextMessage(objectMapper.writeValueAsString(request)));
        } catch (Exception e) {
            log.error("[Error] Failed to send request KIS Websocket - {} / {}", stockCode, e.getMessage());
        }
    }
}

  KisRealTimeTradeWebSocketClient는 한국투자증권 웹소켓 서버로 구독/해제 요청을 보내는 클래스이다. 실제 요청을 보내는 세부 로직은 request() 메서드를 두어 내부에서 처리한다. 외부에는 subscribe(), unsubscribe() 메서드만 노출하여 추상화된 인터페이스를 제공한다.

7.StompInterceptor

@Slf4j
@Component
@RequiredArgsConstructor
public class StompInterceptor implements ChannelInterceptor {
    private final KisSubscriptionManager kisSubscriptionManager;
    private final StockSearchService stockSearchService;

    private final static String STOCK_CODE_HEADER_NAME = "stockCode";

    /**
     * 특정 종목 구독 및 해제 시 한국투자증권 웹소켓 연결 관리를 위한 메서드
     *
     * - 구독 등록 시, 한국투자증권 주식 체결가 웹 소켓 등록 요청
     * - 구독 해제 시, 한국투자증권 주식 체결가 웹 소켓 해제 요청
     *
     * @param message : 수신 메시지
     * @param channel : 메시지 채널
     * @return        : 기본 처리 메서드 호출
     */
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        String stockCode = extractStockCode(accessor);

        if (StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
            try {
                stockSearchService.increaseStockSearchCount(stockCode);
                kisSubscriptionManager.subscribe(stockCode);
            } catch (Exception e) {
                log.error("[ERROR] Failed to subscribe stock {} - {}", stockCode, e.getMessage());
                return null;
            }
            
        }
        
        if (StompCommand.UNSUBSCRIBE.equals(accessor.getCommand())) {
            try {
                kisSubscriptionManager.unsubscribe(stockCode);
            } catch (Exception e) {
                log.error("[ERROR] Failed to unsubscribe stock {} - {}", stockCode, e.getMessage());
                return null;
            }
        }

        return message;
    }
    
    /**
     * STOMP 요청 메시지 헤더 내 주식 종목 코드 추출 메서드
     *
     * @param accessor  STOMP 헤더 접근 객체
     * @return          주식 종목 코드
     */
    private String extractStockCode(StompHeaderAccessor accessor) {
        return accessor.getFirstNativeHeader(STOCK_CODE_HEADER_NAME);
    }
}