Axon Framework (Command와 Event 구현)

저번 글(https://jobc.tistory.com/223)에서 Axon Framework의 전체적인 개념을 간단히 알아봤다면, 이번에는 실제로 구현을 해보며 학습한 기록을 남긴다.

 

Spring Boot와 Axon Framework

Axon Framework는 Spring Boot와 친화적이다.

따라서 다양한 어노테이션을 지원하고, 간단하게 의존성 추가로 Axon를 사용할 수 있다.

implementation("org.axonframework:axon-spring-boot-starter:4.5")

 

Command 구현

메세지 종류 중에 Command 메세지를 발행해보고 핸들러까지 구현해본다.

 

Command Gateway

Command Gateway는 Command 메세지 통로 역할을 수행한다.

@Autowired
private lateinit var commandGateway: CommandGateway

fun sendCommand() {
	commandGateway.send(CommandMessage)
	commandGateway.sendAndWait(CommandMessage)
}

 

- send() : Async

- sendAndWait() : Sync

 

위처럼 간단하게 Command Gateway를 이용한다면 손쉽게 커맨드 메세지를 발행할 수 있다.

 

Command Handler

커맨드 메세지를 발행했으니, 이제 그 메세지를 핸들링할 핸들러가 필요하다.

커맨드 핸들러는 일반 컴포넌트에도 작성할 수 있지만, 상태 변경이 진행되는 Aggregate안에 작성하는 것을 추천한다고 한다.

@Aggregate
class OrderAggregate(
   @AggregateIdentifier val orderId: Int // Aggregate 식별자
) {
   var price: Int = 0

   @CommandHandler
   fun completeOrder(orderCompleteCommand: OrderCompleteCommand) {
      logger.info("OrderCompleteCommand: $orderCompleteCommand")

      // 이벤트가 발행되면 EventBus를 통해 AxonServer EventStore에 저장된다.
      AggregateLifecycle.apply(
         OrderCompleteEvent(orderCompleteCommand.orderId)
      )
   }
}

data class OrderCompleteCommand(
   @TargetAggregateIdentifier
   val orderId: Int
)

커맨드 핸들러에 Event 발행을 작성했다.

보통 커맨드 핸들러에 정말 상태 변경을 진행할 것인지 조건을 넣게된다.

 

Command Bus

커맨드 버스는 커맨드 메세지가 이동하는 통로이다.

인스턴스 내부에서만 커맨드가 이동할 수 있는 종류도 있고, Axon Server를 거치는 커맨드 버스를 구성할 수도 있다.

Axon Starter를 import해서 사용하고 있다면 기본적으로 Axon Server로 연결되는 AxonServerCommandBus가 구성된다.

 

Command 핸들러가 중복된 경우

Command는 목적지가 하나이다. 해당 커맨드 메세지를 수행할 핸들러는 하나여야 된다는 것이다.

그런데 만약 Command에 대한 핸들러가 여러개라면?

Warning을 표시하며, 최근에 등록한 핸들러로 전달된다.

 

Event 구현

Event 발행은 커맨드를 구현할때 살펴봤다. 이벤트는 보통 Aggregate에서 발행된다.

@CommandHandler
fun issueEvent(IssueCardCommand cmd) {
   AggregateLifecycle.apply(new CardIssuedEvent(cmd.cardId, cmd.amount))
}

Aggregate가 아닌 곳에서 Event 발행이 필요한 경우 Event Gateway를 활용한다.

@Autowired
private lateinit var eventGateway: EventGateway

fun dispatchEvent() {
   eventGateway.publish(CardIssuedEvent("cardId", 100, "shopId"))
}

 

Event Handler 구현

이벤트 핸들러는 `@EventHandler` 어노테이션을 이용하여 구현하며, 첫번째 파라미터는 반드시 이벤트 메세지가 되어야 한다.

그 이후 파라미터는 sequence number, timestamp 등 axon에서 제공해주는 다양한 파라미터를 이용할 수 있다.

@EventHandler
fun on(event: EventA) {
   ...
}

 

Event Processor

이벤트 프로세서는 이벤트 과정의 기술적 측면을 처리하는 구성요소이다.

(사실 아직까지 이 컴포넌트의 역할을 제대로 이해못했는데, 이벤트의 경우 단순한 커맨드-핸들러 구조가 아닌걸로 보인다.)

 

이벤트 프로세서는 구독하는 방식과 추적하는 방식 두가지 형태로 제공된다.

구독하는 방식은 이벤트를 발행한 스레드에서 호출되고, 추적하는 방식은 자체적인 스레드에서 메세지를 가져와 관리한다.

기본적으로 Axon은 Tracking Event Processr(추적하는 방식)을 사용한다.

Tracking 방식은 Subscribing(구독하는 방식)과 다르게 그 과정을 저장할 Token 저장소가 필요하다.

Tracking Event Processor가 이벤트 스트림을 통해 받는 각 메세지에는 토큰이 담겨있다.

이 토큰을 사용하면 프로세서가 나중에 스트림을 다시 열여서 마지막 이벤트와 함께 멈췄던 스트림을 다시 시작할 수 있다.

 

이벤트 프로세서는 이벤트 핸들러를 그룹으로 묶어 역할을 수행하는데, 하는 역할은 대략 다음과 같다.

- 이벤트 핸들러 간에 순서 조정

- Error Handling 설정

- Replaying Events

 

지금까지 이해한바로는 Event Processor가 각 이벤트 핸들러간의 공통 기능을 수행해주는 역할로 보인다.

(Spring으로 따지면 Aspect같은 역할?)

 

Event Store

Axon Framework Starter 의존성을 추가하면 기본적으로 Axon Server가 이벤트 저장소 역할을 한다.

저장소에 저장될 때 Serializable 형태를 설정할 수 있는데, 기본적으로 XStreamSerializer가 적용되어있다.

 

Event Sourcing Handler

이벤트 핸들러 중에 EventSourcingHandler도 있다.

EventSourcingHandler는 DDD로 프로젝트를 구성했을 경우 Aggregate 상태를 변경해주는 Handler이다.

DDD와 EventSourcing 패턴에 익숙하지 않아서, 위에서 구현한 Event Handler와 어떤 차이가 있는지 혼란스러웠는데, Event SourcingHandler는 Aggregate 상태 변경을 저장하는데에만 목적이 있는 것 같았다.

반면, Event Handler는 이벤트에 대한 핸들링이고, Aggregate 상태 변경외에 다른 액션을 취하고 싶을 때 사용된다.

 

 

 

이렇게 Command와 Event에 대해 알아보고 구현까지 해봤지만, 실제 운용환경에서는 어떻게 작동할지 감이 안잡힌다.

(분산환경에서 Command가 발행되면 여러 인스턴스 중 어떤 인스턴스의 Command Handler가 반응할 것인가? 등등)

우선 기초적인 개념정도만 이해하고 실제로 사용하게 될 때 충분한 테스트가 필요할 것으로 보인다.

다음 글에선 내가 Axon 을 접하게된 계기인 Axon의 saga pattern에 대해 정리해보고 구현해볼 계획이다.