Architecture Patterns with Python(10장)

커맨드와 커맨드 핸들러

 지난 9장에서 모든 유스 케이스 함수를 이벤트 핸들러로 변경했다. API 는 새 배치를 만드는 POST 를 받으면 새로운 BatchCreated 이벤트를 만들어서 내부 이벤트처럼 처리한다.

 

위의 workflow 는 직관적으로 봤을 때 조금 어색함이 느껴진다. 배치가 생성되지 않은 상태에서 먼저 BatchCreated 라는 이벤트가 생성되고, 그 후에 배치가 진짜로 생성된다.

 

그렇다면 이름을 단순히 CreateBatch 로 바꾸면 될까?

 

여기서 사용되는 개념이 바로 커맨드(command) 라는 개념이다. 이번 챕터에서는 메시지를 다루는 방법 중, 커맨드와 이벤트를 활용하여 기존 코드를 리팩토링 해본다.

 

10.1 커맨드와 이벤트

이벤트와 마찬가지로 커맨드도 메시지의 일종이다. 시스템의 한 부분에서 다른 부분으로 전달되는 명령이 커맨드다. 보통 커맨드를 아무 메서드도 들어있지 않은 데이터 구조로 표현하고 이벤트와 거의 같은 방식으로 처리한다.

 

하지만 분명한 차이가 있다.

 

커맨드는 한 행위자로부터 다른 구체적인 행위자에게 전달된다. 보내는 행위자는 받는 행위자가 커맨드를 받고 구체적인 작업을 수행하길 바란다. API 핸들러에 폼을 전달하는 행위는 커맨드를 전달하는 행위와 같다. 그래서 커맨드의 이름을 붙일 때는 'allocate stock', 'delay shipment' 와 같은 명령형 동사구를 사용한다.

 

커맨드는 의도를 잡아낸다. 커맨드는 시스템이 어떤 일을 수행하길 바라는 의도를 드러낸다. 그 결과로 커맨드를 보내는 행위자는 커맨드 수신자가 커맨드 처리에 실패했을 때 오류 정보를 돌려받기를 바란다.

 

반면 이벤트는 행위자가 관심있는 모든 리스너에게 보내는 메시지다. BatchQuantityChanged 라는 이벤트를 발행해도 발행하는 행위자는 누가 이 이벤트를 받는지에 대해 모른다. 이벤트를 보내는 쪽은 받는 쪽의 성공이나 실패에 관심이 없다.

 

	이벤트	커맨드
이름	과거형	명령형
오류 처리	송신하는 쪽과 독립적으로 실패	송신하는 쪽에 오류를 돌려주면서 실패
받는 행위자	모든 리스너	정해진 수신자

 

from datetime import date
from typing import Optional
from dataclasses import dataclass


class Command:
    pass


@dataclass
class Allocate(Command):  # 1
    orderid: str
    sku: str
    qty: int


@dataclass
class CreateBatch(Command):  # 2
    ref: str
    sku: str
    qty: int
    eta: Optional[date] = None


@dataclass
class ChangeBatchQuantity(Command):  # 3
    ref: str
    qty: int

 

1. commands.Allocate 는 events.AllocationRequired 를 대신한다.

2. commands.CreateBatch 는 events.BatchCreated 를 대신한다.

3. commands.ChangeBatchQuantity 는 events.BatchQuantityChanged 를 대신한다.

 

10.2 예외 처리 방식의 차이점

단순히 이름을 변경하는 것은 아무런 문제가 없다. 그렇게 해도 시스템의 동작이 바뀌지는 않기 때문이다. 하지만 메시지를 취급하는 메시지버스에도 변경이 필요하다.

 

Message = Union[commands.Command, events.Event]


def handle(
    message: Message,
    uow: unit_of_work.AbstractUnitOfWork,
):  # 1
    results = []
    queue = [message]
    while queue:
        message = queue.pop(0)
        if isinstance(message, events.Event):
            handle_event(message, queue, uow)  # 2
        elif isinstance(message, commands.Command):
            cmd_result = handle_command(message, queue, uow)  # 2
            results.append(cmd_result)
        else:
            raise Exception(f"{message} was not an Event or Command")
    return results

 

1. 여전히 메시지를 입력받는 주 진입점이 있다.

2. 두 가지 도우미 함수를 사용해 이벤트와 커맨드를 전달한다.

 

다음은 이벤트에 대한  처리 방법을 보여준다.

 

def handle_event(
    event: events.Event,
    queue: List[Message],
    uow: unit_of_work.AbstractUnitOfWork,
):
    for handler in EVENT_HANDLERS[type(event)]:  # 1
        try:
            logger.debug("handling event %s with handler %s", event, handler)
            handler(event, uow=uow)
            queue.extend(uow.collect_new_events())
        except Exception:
            logger.exception("Exception handling event %s", event)
            continue  # 2

 

1. 한 이벤트를 여러 핸들러가 처리하도록 위임할 수 있는 디스패처로 이벤트가 처리된다.

2. 오류가 발생하면 오류를 찾아서 로그에 남기지만, 오류가 다른 메시지 처리를 방해하지는 못하게 한다.

 

다음은 커맨드 처리 방법이다.

 

def handle_command(
    command: commands.Command,
    queue: List[Message],
    uow: unit_of_work.AbstractUnitOfWork,
):
    logger.debug("handling command %s", command)
    try:
        handler = COMMAND_HANDLERS[type(command)]  # 1
        result = handler(command, uow=uow)
        queue.extend(uow.collect_new_events())
        return result  # 3
    except Exception:
        logger.exception("Exception handling command %s", command)
        raise  # 2

 

1. 커맨드 디스패처는 커맨드 한 개에 핸들러 한 개만 허용한다.

2. 오류가 발생하면 빠르게 실패하면서 오류를 위로 전파한다.

3. return result 는 임시방편이다. API 에 배치에 대한 참조를 돌려주기 위해 임시로 채택했다. 이 문제에 대한 수정은 12장에서 한다.

 

위의 이벤트와 커맨드 함수에서 각각의 이벤트와 커맨드를 디스패치하기 위해 사용하는 HANDLERS 사전을 나눈다.

앞서 정의한 바에 따르면, 한 커맨드에는 핸들러가 하나밖에 없다.

 

EVENT_HANDLERS = {
    events.OutOfStock: [handlers.send_out_of_stock_notification],
}  # type: Dict[Type[events.Event], List[Callable]]

COMMAND_HANDLERS = {
    commands.Allocate: handlers.allocate,
    commands.CreateBatch: handlers.add_batch,
    commands.ChangeBatchQuantity: handlers.change_batch_quantity,
}  # type: Dict[Type[commands.Command], Callable]

 

10.3 이벤트, 커맨드, 오류 처리

이벤트는 실패 로그만 남기고 행위를 지속하는데, 만약 이벤트 처리에 실패하면 시스템의 일관성은 유지될 수 있는가?

 

답은 그렇다.

 

그 이유는 애그리게이트를 통해 일관성 경계로 동작하게 했고, 애그리게이트에 대한 업데이트 성공이나 실패를 원자적으로 처리하기 위해 UoW 를 설계했기 때문이다.

 

예를 들어 주문에 재고를 할당하면 일관성 경계는 Product 애그리게이트다. 즉 실수로 과할당할 수 없다는 뜻이다. 애그리게이트는 도메인 객체 컬랙션을 한 번에 다룰 수 있게 해주는 객체이며, 객체를 변경하기 위한 유일한 진입점이다. 애그리게이트는 비즈니스 로직에 반하는 상황에 대하여 이를 거부한다. 또한 UoW 를 통해 데이터의 무결성을 유지한다. (7장 참고)

 

사용자가 시스템이 어떤 일을 하기를 원한다면 이 요청을 커맨드로 표현한다. 커맨드는 한 애그리게이트를 변경해야 하고, 전체적으로 성공하거나 전체적으로 실패해야 한다. 시스템이 수행하는 다른 북키핑, 정리, 통지는 이벤트를 통해 발생한다. 커맨드가 성공하기 위해 이벤트 핸들러가 성공할 필요는 없다.

 

이벤트 기반 시스템에서는 애그리게이트가 상태를 데이터베이스에 영속화한 다음에 이벤트를 발생시킨다. 상태를 영속화하기 전에 이벤트를 발생시키고 동시에 모든 변화를 커밋하면 어떤 일이 발생할까?

 

이메일을 보내는 이벤트가 있다고 할 때, 이메일 서버가 과부하 상태라면 어떨까? 모든 작업이 동시에 완료되어야 한다면 바쁜 이메일 서버가 주문을 결제하는 과정을 방해할 수도 있다.

 

이런 관심사를 분리하면 실패할 수 있는 요소들이 서로 격리되어 실패하게 할 수 있다. 이렇게 하면 시스템 전체의 신뢰성이 높아진다. 가장 중요한 부분며 반드시 성공해야 하는 부분은 고객이 신경 쓰는 유일한 부분인 주문에 대한 커맨드다.

 

10.4 동기적으로 오류 복구하기

불가피하게 이벤트에서 오류가 발생한 경우 이를 복구시키기 위해서는 어떻게 해야 할까?

 

가장 먼저 언제 오류가 났는지 로그를 사용하여 확인하는 것이다. 

 

def handle_event(
    event: events.Event,
    queue: List[Message],
    uow: unit_of_work.AbstractUnitOfWork,
):
    for handler in EVENT_HANDLERS[type(event)]:
        try:
            logger.debug("handling event %s with handler %s", event, handler)
            handler(event, uow=uow)
            queue.extend(uow.collect_new_events())
        except Exception:
            logger.exception("Exception handling event %s", event)
            continue

 

로그에 저장된 데이터를 사용해 문제를 단위 테스트로 재현하거나 시스템에서 메시지를 다시 실행할 수 있다.

 

버그 외에도 시스템은 언제나 백그라운드에서 일시적인 실패가 발생할 수 있다. 네트워크 문제, 데이터베이스 테이블의 교착상태, 배치로 인해 발생하는 일시 중단 등이 있다.

 

이런 경우 재시도를 통해 시스템 상태를 우아하게 복구할 수 있다. 우아하게 라는 부사가 의미하는 것은 처음에는 성공하지 못했지만 지수적으로 증가되는 back-off 시간 이후에 연산을 재시도 한다는 뜻이다.

 

from tenacity import Retrying, RetryError, stop_after_attempt, wait_exponential

def handle_event(
    event: events.Event,
    queue: List[Message],
    uow: unit_of_work.AbstractUnitOfWork,
):
    for handler in EVENT_HANDLERS[type(event)]:
        try:
            for attempt in Retrying(
                stop=stop_after_attempt(3),
                wait=wait_exponential()
            ):
                with attempt:
                    logger.debug("handling event %s with handler %s", event, handler)
                    handler(event, uow=uow)
                    queue.extend(uow.collect_new_events())
        except RetryError as retry_failure:
            logger.error(
                'Failed to handle event %s times, giving up!,
                retry_failure.last_attempt.attempt_number
            )
            continue

 

위와 같이 파이썬의 tenacity 는 재시도와 관련된 일반 패턴을 구현한다. 여기서는 메시지 버스가 연산을 세 번 재시도하도록 설정했다.

 

실패할 수도 있는 연산을 재시도하는 것은 시스템의 회복 탄력성을 향상시키는 최선의 방식을 것이다. 여기서도 작업 단위와 명령 핸들러 패턴이 각 재시도가 일관성 있는 시스템 상태를 보장하고, 시스템이 작업이 반쯤 끝난 상태로 남지 않게 해준다.

 

참고) tenacity 와 관계없이 어느 시점에는 메시지를 처리하는 시도를 포기해야 한다. 자세한 내용은 14장을 참고

 

10.5 마치며

장점	단점
커맨드와 이벤트를 다른 방식으로 처리하면 어떤 부분이 꼭 성공해야 하는지, 어떤 부분은 나중에 정리해도 되는지를 구분하는 데 도움이 된다.	커맨드와 이벤트의 의미적인 차이가 애매하다. 이로 인해 둘 사이의 차이에 대해 고민하는 데 시간을 낭비할 수 있다.
CreateBatch 는 BatchCreated 라는 이름보다 덜 혼란스럽다. 명시성은 암시성보다 낫다.	실패를 명확히 구분한다. 때로는 프로그램이 깨질 수 있다는 사실을 알고, 실패를 더 작게 만들고 격리시키는 방식으로 처리한다. 이로 인해 시스템에 대해 추론하기가 더 어려워질 수 있고 모니터링을 더 잘 해야 할 필요가 생긴다.