[마이크로서비스 패턴 쉽게 개발 4] Axon Framework(2): CommandGateway와 Event Sourcing

Micro Service/mSVC개발

[마이크로서비스 패턴 쉽게 개발 4] Axon Framework(2): CommandGateway와 Event Sourcing

Happy@Cloud 2024. 1. 18. 22:42

이번 편에서는 10번까지 개발 해 보겠습니다.

ElephantService 완성

2. Check Validation

비즈니스 요건이 아래와 같을 때 Command 발송 전에 validation 체크를 먼저 합니다.

코끼리 등록 시 몸무게는 30kg ~ 200kg 사이만 허용한다.
이미 냉장고에 들어가 있는 코끼리는 또 넣을 수 없다.
냉장고 안에 들어가 있는 코끼리만 꺼낼 수 있다.

- create 메소드

    public ResultVO<CreateElephantCommand> create(ElephantDTO elephant) {
        log.info("[ElephantService] Executing create: {}", elephant.toString());

        ResultVO<CreateElephantCommand> retVo = new ResultVO<>();

        //check validation
        if(elephant.getWeight() < 30 || elephant.getWeight() > 200) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage("몸무게는 30kg이상 200kb이하로 입력해 주세요.");
            return retVo;
        }
        
        return retVo;
    }

- enter 메소드

코끼리의 현재 상태를 읽어야 하므로 JpaRepository를 확장한 ElephantRepository가 필요 합니다.

@Autowired 어노테이션을 이용하여 Spring Bean class로 등록된 ElephantRepository class의 객체인 elephantRepository를 생성 합니다. Bean class를 Wiring 하는 방법은 아래와 같이 생성자 메소드를 이용하는 방법으로 하는게 제일 좋습니다.

public class ElephantService {

    private final ElephantRepository elephantRepository;
    @Autowired
    public ElephantService(ElephantRepository elephantRepository) {
        this.elephantRepository = elephantRepository;
    }
...
}

냉장고 안에 있는 코끼리인지 체크 합니다.

    public ResultVO<String> enter(String id) {
        log.info("[ElephantService] Executing enter for Id: {}", id);
        ResultVO<String> retVo = new ResultVO<>();

        //-- check validation
        Elephant elephant = getEntity(id);
        if(elephant.getStatus().equals(StatusEnum.ENTER.value())) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage("이미 냉장고 안에 있는 코끼리입니다.");
            return retVo;
        }

        return retVo;
    }

id에 해당하는 코끼리 데이터를 리턴하는 getEntity 함수를 만듭니다.

JPA에서 제공하는 findById를 이용하여 데이터를 읽습니다.

    private Elephant getEntity(String id) {
        Optional<Elephant> optElephant = elephantRepository.findById(id);
        return optElephant.orElse(null);
    }

코끼리 상태 코드를 위한 StatusEnum class를 만듭니다.

package 'org.axon.dto' 하위에 만듭니다. 'Enumeration(열거)'형으로 만듭니다.

public enum StatusEnum {
    READY("Ready"),
    ENTER("Enter"),
    EXIT("Exit");

    private String value;
    StatusEnum(String value) { this.value = value; }
    public String value() { return this.value; }
}

- exit 메소드

현재 코끼리 상태가 냉장고 안에 있는지 체크하여 validation 검사를 합니다.

    public ResultVO<String> exit(String id) {
        log.info("[ElephantService] Executing exit for Id: {}", id);
        ResultVO<String> retVo = new ResultVO<>();

        //-- check validation
        Elephant elephant = getEntity(id);
        if(!elephant.getStatus().equals(StatusEnum.ENTER.value())) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage("냉장고 안에 있는 코끼리만 꺼낼 수 있습니다.");
            return retVo;
        }

        return retVo;
    }

3. Command 발송

AxonServer에 Command를 발송하기 위해서는 Axon Framework의 CommandGateway 객체가 필요 합니다.

CommandGateway객체는 일단 필드를 직접 Auto wired하여 생성 합니다. 뒤쪽에서 생성자에서 Auto wired 하도록 변경 하겠습니다.

public class ElephantService {
    @Autowired
    private transient CommandGateway commandGateway;
...
}

중요) Axon Framework에서 메시지를 송/수신하는 class들은 반드시 'transient' 키워드를 붙여야 함
'transient' 키워드의 의미는 통신되는 메시지를 Binary로 변환하지 않고 원문 그대로 보낸다는 의미입니다.
CommandGateway, EventGateway, QueryGateway의 객체를 생성할 때는 반드시 transient 키워드를 붙여야 합니다.

- 코끼리 생성 요청 Command를 보냅니다.

id 값은 랜덤하게 3자리로 부여 합니다.
name과 weight는 Swagger page에서 입력하여 APIController로 부터 받은 값을 보냅니다.
status는 'Ready' 상태로 요청 합니다.

CommandGateway의 메소드에는 대표적으로 2개가 있습니다.

sendAndWait: Blocking 방식으로 보냅니다. Command Handler가 수신할 때까지 기다립니다. 최대 기다리는 시간은 뒤쪽에 지정된 시간입니다. 예제에서는 30초 입니다.
send: Non-Blocking 방식으로 보냅니다. 보내고 바로 다음 라인 처리를 합니다. callback 함수를 만들 수도 있습니다.

public ResultVO<CreateElephantCommand> create(ElephantDTO elephant) {
...
        //send command
        CreateElephantCommand cmd = CreateElephantCommand.builder()
                .id(RandomStringUtils.random(3, false, true))
                .name(elephant.getName())
                .weight(elephant.getWeight())
                .status(StatusEnum.READY.value())
                .build();

        try {
            commandGateway.sendAndWait(cmd, 30, TimeUnit.SECONDS);
            retVo.setReturnCode(true);
            retVo.setReturnMessage("Success to create elephant");
            retVo.setResult(cmd);
        } catch(Exception e) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage(e.getMessage());
        }
        
        return retVo;
 }

'CreateElephantCommand' class는 이미 package 'org.axon.command'밑에 만들어져 있습니다.

- 코끼리 넣기 요청 Command도 보냅니다.

'EnterElephantCommand'의 객체를 생성하여 보냅니다.

    public ResultVO<String> exit(String id) {
...
        //--send command
        try {
            commandGateway.sendAndWait(EnterElephantCommand.builder()
                    .id(id)
                    .status(StatusEnum.READY.value())
                    .build(), 30, TimeUnit.SECONDS);
            
            retVo.setReturnCode(true);
            retVo.setReturnMessage("Success to request enter elephant");
        } catch(Exception e) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage(e.getMessage());
        }

        return retVo;
    }

package 'org.axon.command' 하위에 'EnterElephantCommand' class를 만듭니다.

위 enter메소드에서 이 class의 객체를 생성하면서 id에는 코끼리ID를 넣고 status에는 'Ready'값을 넣게 됩니다.

import lombok.Builder;
import lombok.Value;
import org.axonframework.modelling.command.TargetAggregateIdentifier;

@Value
@Builder
public class EnterElephantCommand {
    @TargetAggregateIdentifier
    String id;
    String status;
}

중요) @TargetAggregateIdentifier를 반드시 지정하세요.
Command는 Axon Server를 통해 ElephantAggregate로 전달이 됩니다.
ElephantAggregate는 어떤 코끼리 객체에 대해 Event Replay를 할 지 알아야 합니다.
따라서 각 코끼리 객체를 구별할 수 있는 Unique key인 'id'를 @TargetAggregateIdentifier 어노테이션으로 지정해 줘야 합니다.

- 코끼리 꺼내기 Command 발송을 추가 합니다.

넣기 Command와 비슷하게 'ExitElephantCommand' 객체를 생성하여 보냅니다.

    public ResultVO<String> exit(String id) {
...
        //-- send command
        try {
            commandGateway.sendAndWait(ExitElephantCommand.builder()
                    .id(id)
                    .status(StatusEnum.EXIT.value())
                    .build(), 30, TimeUnit.SECONDS);

            retVo.setReturnCode(true);
            retVo.setReturnMessage("Success to request exit elephant");
        } catch(Exception e) {
            retVo.setReturnCode(false);
            retVo.setReturnMessage(e.getMessage());
        }

        return retVo;
    }

package 'org.axon.command' 하위에 'ExitElephantCommand' class도 만듭니다.

import lombok.Builder;
import lombok.Value;
import org.axonframework.modelling.command.TargetAggregateIdentifier;

@Value
@Builder
public class ExitElephantCommand {
    @TargetAggregateIdentifier
    String id;
    String status;
}

4. Return(요청완료)

CommandGateway를 통해 Axon Server로 발송이 완료되면 그 처리 결과를 APIController로 리턴하게 되고 최종적으로 Swagger page에서 그 응답을 확인할 수 있게 됩니다.

오해하지 말아야 할 것은 commandGeteway.sendAndWait로 요청한 후 Command 요청이 모두 처리되고 나서 리턴되는게 아니라는 겁니다. 즉, 위 프로세스에서 10번까지 완료하고 리턴되는게 아니라는 뜻입니다.

5번 AxonServer가 CommandHandler인 ElephantAggregate로 Command를 Push하면 리턴됩니다. 그 이후의 프로세스는 별개로 동작합니다.

ElephantAggregate 개발

5번에서 9번까지를 담당하는 ElephantAggregate를 개발 합니다.

5. Push Command

Axon 서버는 연결된 어플리케이션 중 Command를 처리할 수 있는 Command Handler가 있는 어플리케이션을 찾아

Command 메시지를 송부 합니다.

Swagger Page에서 코끼리 생성 API를 테스트 해 보십시오.

아래와 같이 Command Handler를 찾을 수 없다는 결과가 리턴 됩니다.

6. Event Replay

Command Handler의 역할을 하는 Aggregate 'ElephantAggregate'를 만들겠습니다.

package 'org.axon.aggregate'를 만들고 하위에 'ElephantAggregate' class를 작성합니다.

@Aggregate: Aggregate 역할을 하는 class라는 의미입니다.
Aggregate가 관리할 필드 구조를 정의합니다. 이 필드들이 Event Replay로 구할 최종 데이터들이고 Event Store에 저장될 데이터가 됩니다.
- @AggregateIdentifier: 각 Entity를 구별하는 Primary key 필드입니다.
- @AggregateMember: Entity의 멤버 필드입니다.
ElephantAggregate() : 인자가 없고 내용도 없는 빈 생성자입니다. Axon Framework에서 내부적으로 필요하기 때문에 반드시 만들어 줘야 합니다.

import lombok.extern.slf4j.Slf4j;
import org.axonframework.modelling.command.AggregateIdentifier;
import org.axonframework.modelling.command.AggregateMember;
import org.axonframework.spring.stereotype.Aggregate;

@Slf4j
@Aggregate
public class ElephantAggregate {
    @AggregateIdentifier
    private String id;
    @AggregateMember
    private String name;
    @AggregateMember
    private int weight;
    @AggregateMember
    private String status;

    public ElephantAggregate() {}

}

최초 등록이기 때문에 지금은 Replay할 Event가 없으니 다음 단계로 넘어 갑니다.

7. Biz Logic 처리 & 8. Event 생성

예제에서는 특별히 처리할 Biz Logic이 없기 때문에 Command Handler에서는 Event만 생성 합니다.

- 코끼리 생성 CommandHandler 작성

@CommandHandler: Command Handler class라는 의미입니다.
ElephantAggregate(CreateElephantCommand cmd): 새로운 Data Entity생성 처리는 Aggregate class의 생성자 메소드에서 합니다. 파라미터로는 Axon Server로 푸시 받은 Command 메시지가 들어 옵니다.
CreatedElephantEvent event ... : Command 요청을 처리한 후 완료되었다는 Event를 생성 합니다.
AggregateLifeCycle.apply(event): 생성한 Event를 발행합니다. 이렇게 발행된 Event는 Axon 서버에 보내지는게 아닙니다. 따라서 이 Event를 이용할 수 있는 범위는 현재 어플리케이션 내부입니다.

...
    //-- 새로운 코끼리 객체 생성 처리
    @CommandHandler
    private ElephantAggregate(CreateElephantCommand cmd) {
        log.info("[@CommandHandler] CreateElephantCommand for Id: {}", cmd.getId());
        CreatedElephantEvent event = new CreatedElephantEvent(
                cmd.getId(), cmd.getName(), cmd.getWeight(), cmd.getStatus());
        AggregateLifecycle.apply(event);
    }

중요) Command 이름은 현재형이고 Event 이름은 과거형으로 짓는 이유
DDD의 영향으로 볼 수 있습니다.
DDD 전략 설계 방법 중 Event Storming에서는 Command는 Event를 발생 시키는 요청 Action이기 때문에 현재형으로 표현하고,
Event는 발생한 결과이므로 과거형으로 표현하도록 안내하고 있습니다.
이 가이드를 준용하기 위해 보통 Command는 현재형으로 Event는 과거형으로 짓습니다.
https://happycloud-lee.tistory.com/94

package 'org.axon.events'를 만들고 하위에 Event class 'CreatedElephantEvent'를 작성 합니다.

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class CreatedElephantEvent {
    private String id;
    private String name;
    private int weight;
    private String status;
}

TIP) Event class를 만들때 Command class를 복사-붙여넣기 하여 만드세요.
Command class와 Event class의 필드 구조는 대부분 동일합니다.
Command객체에서 요청된 데이터로 Event객체를 생성하여 Event 저장소에 저장하거나 조회DB에 반영하여야 하므로 당연한 동일한 것입니다. 따라서 class도 복사-붙여넣기 하여 만드는게 편리하고 실수를 방지할 수 있습니다.
주의) Event class에는 @TargetAggregateIdentifier를 붙이시면 안됩니다.

- 코끼리 넣기 Command Handler 작성

Command Handler를 아래와 같이 작성합니다.

    //-- 냉장고에 넣기
    @CommandHandler
    private void handle(EnterElephantCommand cmd) {
        log.info("[@CommandHandler] EnterElephantCommand for Id: {}", cmd.getId());

        AggregateLifecycle.apply(new EnteredElephantEvent(cmd.getId(), cmd.getStatus()));
    }

중요) 새로운 Entity 생성시에만 생성자 메소드를 사용해야 함
새로운 Aggregate의 Entity를 처리할 때는 생성자 메소드를 이용합니다.
하지만 Entity의 일부 필드값만 변경하는 Command Handler는 생성자 메소드로 작성하면 절대 안됩니다.
또 다른 Entity가 만들어져서 이상하게 동작합니다.
보통 'handle' 이라는 이름의 메소드에 작성합니다.

package 'org.axon.events'밑에 Event class 'EnteredElephantEvent'를 만듭니다.

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class EnteredElephantEvent {
    private String id;
    private String status;
}

- 코끼리 꺼내기 CommandHandler 작성

Command Handler를 아래와 같이 작성합니다.

    //-- 냉장고에서 꺼내기
    @CommandHandler
    private void handle(ExitElephantCommand cmd) {
        log.info("[@CommandHandler] ExitElephantCommand for Id: {}", cmd.getId());

        AggregateLifecycle.apply(new ExitedElephantEvent(cmd.getId(), cmd.getStatus()));
    }

package 'org.axon.events' 밑에 Event class 'ExitedElephantEvent'도 만듭니다.

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class ExitedElephantEvent {
    private String id;
    private String status;
}

9. Event 추가

생성/발행된 Event 메시지를 Event Store에 추가 합니다.

Event Store에 추가하는 메소드는 @EventSourcingHandler 어노테이션을 지정 합니다.

- 생성된 코끼리 Event 데이터 추가

최초 등록이므로 Aggregate의 모든 필드에 값을 셋팅합니다.

    @EventSourcingHandler
    private void on(CreatedElephantEvent event) {
        log.info("[@EventSourcingHandler] CreatedElephantEvent for Id: {}", event.getId());
        this.id = event.getId();
        this.name = event.getName();
        this.weight = event.getWeight();
        this.status = event.getStatus();
    }

- 냉장고 안에 넣어진 코끼리 상태 Event 추가

확인을 위해 이전/이후 코끼리 상태를 로그에 찍습니다.

현재 코끼리 ID에 해당하는 Entity에서 변경되는 값인 'status' 필드 값만 변경 합니다.

    @EventSourcingHandler
    private void on(EnteredElephantEvent event) {
        log.info("[@EventSourcingHandler] EnteredElephantEvent for Id: {}", event.getId());
        log.info("======== [넣기] Event Replay => 코끼리 상태: {}", this.status);

        this.status = event.getStatus();

        log.info("======== [넣기] 최종 코끼리 상태: {}", this.status);

    }

- 냉장고에서 꺼내어진 코끼리 상태 Event 추가

역시 변경되는 'status' 필드 값만 업데이트 합니다.

    @EventSourcingHandler
    private void on(ExitedElephantEvent event) {
        log.info("[@EventSourcingHandler] ExitedElephantEvent for Id: {}", event.getId());
        log.info("======== [꺼내기] Event Replay => 코끼리 상태: {}", this.status);

        this.status = event.getStatus();

        log.info("======== [꺼내기] 최종 코끼리 상태: {}", this.status);

    }

ElephantEventHandler 작성

이제 마지막 단계인 '10번 코끼리 데이터 생성, 상태 변경'을 개발할 차례입니다.

이 과정은 조회(Query)를 위해 조회DB에 최종 데이터 상태를 저장하는 것입니다.

만약 조회 전용 DB를 다른 마이크로서비스에서 관리한다면 이 과정은 없어도 됩니다.

하지만 Event Sourcing 패턴을 적용할 때는 조회 속도 향상을 위해 조회 DB 업데이트도 반드시 해야 합니다.

왜냐하면 Event Sourcing 패턴을 적용할때는 데이터의 최종 상태를 Event Replay를 통해 계산하기 때문에

조회 요청에 빠르게 응답할 수 없기 때문입니다.

package 'org.axon.events' 밑에 'ElephantEventHandler' class를 작성합니다.

소스를 보시면 아시겠지만 JPA를 이용하여 DB에 Write합니다.

import lombok.extern.slf4j.Slf4j;
import org.axon.entity.Elephant;
import org.axon.repository.ElephantRepository;
import org.axonframework.eventhandling.EventHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Slf4j
@Component
public class ElephantEventHandler {
    @Autowired
    private ElephantRepository elephantRepository;

    @EventHandler
    private void on(CreatedElephantEvent event) {
        log.info("[@EventHandler] CreatedElephantEvent for Id: {}", event.getId());
        Elephant elephant = new Elephant();
        elephant.setId(event.getId());
        elephant.setName(event.getName());
        elephant.setWeight(event.getWeight());
        elephant.setStatus(event.getStatus());

        try {
            elephantRepository.save(elephant);
        } catch(Exception e) {
            log.info(e.getMessage());
        }
    }

    @EventHandler
    private void on(EnteredElephantEvent event) {
        log.info("[@EventHandler] EnteredElephantEvent for Id: {}", event.getId());

        Elephant elephant = getEntity(event.getId());
        if(elephant != null) {
            elephant.setStatus(event.getStatus());
            elephantRepository.save(elephant);
        }
    }

    @EventHandler
    private void on(ExitedElephantEvent event) {
        log.info("[@EventHandler] ExitedElephantEvent for Id: {}", event.getId());
        Elephant elephant = getEntity(event.getId());
        if(elephant != null) {
            elephant.setStatus(event.getStatus());
            elephantRepository.save(elephant);
        }
    }

    private Elephant getEntity(String id) {
        Optional<Elephant> optElephant = elephantRepository.findById(id);
        return optElephant.isPresent() ? optElephant.get() : null;
    }
}