[파이썬으로 살펴보는 아키텍처 패턴] 챕터 12 명령-질의 책임 분리(CQRS)

읽기(질의)와 쓰기(명령)는 다르기 때문에 다르게 취급해야 한다 

쓰기 위해 존재하는 도메인 모델

• 지금까지 책에서 도메인 규칙을 강화하는 소프트웨어를 만드는 방법을 설명하기 위해 많은 시간과 노력을 들였음
• 제약과 규칙을 정하고 이를 테스트하고, 일관성 보장을 위해 작업단위나 애그리게이트 같은 패턴 도입 
• 이 모든 복잡도는 시스템 상태 변경 즉 데이터를 유연하게 쓰기 위함 > 그렇다면 읽기는? 

가구를 구매하지 않는 사용자

• 지금까지의 복잡한 도메인 아키텍처 패턴은 쓰기에서는 도움이 되지만 데이터를 읽는데는 아무 역할도 하지 않음
• 재고 할당보다 제품 뷰 건수가 훨씬 많음 -> 쓰기보다 읽기 작업이 훨씬 많다 
• 쓰기 작업은 일관성이 있게 철저하게 유지해야하지만 읽기는 완벽히 최신일 필요는 없다 (캐시나 읽기 전용 복제본 사용가능) 
• 읽기/쓰기를 분리하고 읽기에서 최종 일관성 있게 유지하여 읽기 성능을 향상 할 수 있다 

	읽기	쓰기
동작	간단한 읽기	복잡한 비즈니스 로직
캐시 가능성	높음	캐시 불가
일관성	오래된 값 제공 가능	일관성있는 트랜잭션

 

읽기 일관성을 정말로 달성할 수 있을까?

일관성과 성능을 교환한다는 것은 많은 개발자를 불안하게 만듦 

상품 페이지를 표시한 순간부터 표시된 데이터는 최신 정보가 아니다 
- ex) 상품 페이지 켜놓고 잠시 뒤에 주문했는데 그 사이에 제품이 품절된 경우

모든 분산 시스템은 일관성이 없어서 재고를 할당하려면 반드시 시스템의 현상태를 검사해야함

완전히 일관성을 보장할 수 있다고 해도 갑자기 천재지변이 일어나 상품을 배송못하게 되는 경우도 생김 

현실은 소프트웨어 시스템과 일관성이 없기 때문에 근본적으로 일관성이 없는 데이터를 피할 수 없고, 읽기 측면에서 성능과 일관성을 바꿔도 좋다는 결론 

Post/리디렉션/Get과 CQS(명령-질의 분리)

Post/리디렉션/Get 패턴으로 보는 CQS(Command-Query Seperation) 의 간단한 예 

• /batches 에 POST를 해서 새로운 배치를 만들면 사용자를 batches/123으로 리디렉션해서 새로운 배치를 보여줌 
• 클라이언트는 리디렉션한 url로 자동으로 GET 요청을 보냄 
• POST 요청 페이지를 새로고침하거나 북마크 할때 중복 데이터 제출 방지 
• -> 연산의 쓰기와 읽기 단계를 분리해서 문제 해결 

allocate 엔드포인트 메서드에서 OK와 배치 ID 반환하던 코드에서, OK만 반환하고 새로운 읽기 전용 엔드포인트를 제공해 여기서 할당 상태를 가져오도록 수정 go 

기존 flask_app 

# src/allocate/entrypoints/flask_app.py

@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    try:
        cmd = commands.Allocate(
            request.json["orderid"], request.json["sku"], request.json["qty"]
        )
        uow = unit_of_work.SqlAlchemyUnitOfWork()
        results = messagebus.handle(cmd, uow) # 메시지 버스에서 result 받아서 반환함
        batchref = results.pop(0) 
    except InvalidSku as e:
        return {"message": str(e)}, 400

    return {"batchref": batchref}, 201

수정된 flask_app

from allocation import views

@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    try:
        cmd = commands.Allocate(
            request.json["orderid"], request.json["sku"], request.json["qty"]
        )
        uow = unit_of_work.SqlAlchemyUnitOfWork()
        messagebus.handle(cmd, uow) # 메시지 버스는 반환 값이 없고 그저 커맨드만 보냄 
    except InvalidSku as e:
        return {"message": str(e)}, 400

    return "OK", 202 # 성공시 OK 반환 


@app.route("/allocations/<orderid>", methods=["GET"])
def allocations_view_endpoint(orderid):
    uow = unit_of_work.SqlAlchemyUnitOfWork() 
    result = views.allocations(orderid, uow) # 읽기 쿼리를 날리고 결과 값 받아서 반환
    if not result:
        return "not found", 404
    return jsonify(result), 200

view.py 라는 읽기 전용 뷰를 만들어서 사용 

점심을 잠깐 미뤄라

# src/allocation/views.py

from allocation.service_layer import unit_of_work


def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
    with uow:  ## 하드코딩된 SQL을 사용 한다면? 
        results = list(uow.session/execute(
            'SELECT ol.sku, b.reference'
            ' FROM allocations AS a'
            ' JOIN batches AS b ON a.batch_id = b.id'
            ' JOIN order_lines AS ol ON a.orderline_id = ol.id'
            ' WHERE orderid = :orderid',
            dict(orderid=orderid)
        ))
    return [{'sku': sku, 'batchred': batchref} for sku, batchref in results]

저장소 패턴은 어떻게 하고 그냥 SQL 사용? 

CQRS 뷰 테스트하기

대안을 살펴보기 전에 테스트 코드 먼저 보기 

def test_allocations_view(sqlite_session_factory):
    uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory)
    messagebus.handle(commands.CreateBatch("sku1batch", "sku1", 50, None), uow)
    messagebus.handle(commands.CreateBatch("sku2batch", "sku2", 50, today), uow)
    messagebus.handle(commands.Allocate("order1", "sku1", 20), uow)
    messagebus.handle(commands.Allocate("order1", "sku2", 20), uow)
    # 제대로 데이터를 얻는지 보기 위해 여러 배치와 주문 추가
    messagebus.handle(commands.CreateBatch("sku1batch-later", "sku1", 50, today), uow)
    messagebus.handle(commands.Allocate("otherorder", "sku1", 30), uow)
    messagebus.handle(commands.Allocate("otherorder", "sku2", 10), uow)

	# order1에 대해 잘 할당 됐는지 테스트
    assert views.allocations("order1", uow) == 
        {"sku": "sku1", "batchref": "sku1batch"},
        {"sku": "sku2", "batchref": "sku2batch"},
    ]

 

명확한 대안 1: 기존 저장소 사용하기

도우미 메서드를 product 저장소에 추가하면 어떨까?

# src/allocation/views.py

from allocation.service_layer import unit_of_work


def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
    with uow:  
    # 저장소는 product 객체를 반환해 주어진 주문에서 SKU에 해당하는 모든 상품 찾아야함 
    	products = uow.products.for_order(orderid=orderid) 
    # 모든 배치를 가져옴 
        batches = [b for p in products for b in p.batches]

    return [ # 원하는 주문에 대한 배치만 찾기 위해 배치를 다시 걸러냄 
    	{'sku': b.sku, 'batchref': b.reference} 
        for b in batches 
        if orderid in b.orderids 
    ]

문제점 

• 저장소에 가서 for_order 도 구현해야함 
• Batch 클래스에 가서 orderids 로 구현해야함 
• 이렇듯 도우미 메서드를 양쪽에 다 구차하고 루프문을 돌려야함 
• -> 앞서 말했듯 도메인 모델은 읽기 연산에 최적화 되지 않아서, 도메인 모델 복잡도가 커질수록 읽기연산에 모델 사용하는게 더 어려워짐 

명확한 대안 2: ORM 사용하기

# src/allocation/views.py

from allocation.service_layer import unit_of_work


def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
    with uow: # orm 사용 
        batches = uow.session.query(model.Batch).join(
        	model.OrderLine, model.Batch._allocations
        ).filter(
        	model.OrderLine.orderid == orderid
        )

    return [  
    	{'sku': b.sku, 'batchref': b.batchref} 
        for b in batches 
    ]

• 이 코드가 SQL을 그냥 사용하는 코드보다 실제로 더 쉽게 이해하고 작성 할 수 있을까? 
• ORM을 사용하려면 몇번의 시도가 필요하고 SQLAlchmey 문서를 엄청 많이 봐야한다.  SQL은 SQL일 뿐이다. 
• 게다가 ORM은 몇가지 성능상 문제를 야기함 

고려사항 

ORM에서의 SELECT N+1 문제 

• 객체 리스트를 가져올 때, ORM은 보통 필요한 모든 객체의 ID를 가져오는 질의를 먼저 수행한 후 개별 질의 수행 

성능상 문제 

• 완전히 정규화된 테이블은 쓰기 연산이 데이터 오염을 발생하지 않도록 보장하는 좋은 방법이지만, 데이터를 읽을 때 수많은 join 연산으로 느려질 수 있음 
• 정규화되지 않은 뷰를 추가하거나, 읽기 전용 복사본을 만들거나, 캐시 계층을 추가하면 좋음 

이제는 상어를 완전히 뛰어 넘을때이다 

• 정규화하지 않은 읽기 전용 뷰 테이블을 만들어서 사용
• 잘 튜닝한 인덱스가 있어도 관계형 데이터베이스는 조인을 위해 아주 많은 CPU를 사용하기 때문에 읽기 연산에 최적화된 복사본 생성 
• 데이터를 읽을 때는 동시 연산을 실행하는 클라이언트 수에 제한이 없기 때문에 수평규모 확장이 가능 

from allocation.service_layer import unit_of_work


def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
    with uow:
        results = uow.session.execute(
            """ # 가장 빠른 질의문으로 변경됨 
            SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid
            """,
            dict(orderid=orderid),
        )
    return [dict(r) for r in results]

# src/allocation/adapters/orm.py

allocations_view = Table(
    "allocations_view",
    metadata,
    Column("orderid", String(255)),
    Column("sku", String(255)),
    Column("batchref", String(255)),
)

이벤트 핸들러를 사용해 읽기 모델 테이블 업데이트하기 

읽기 모델을 최신상태로 유지하기 위해 이벤트 아키텍처를 재사용해보자 

# src/allocation/service_layer/messagebus.py

# Allocated 에 대한 새 핸들러 추가 
EVENT_HANDLERS = {
    events.Allocated: [ # 할당 
        handlers.publish_allocated_event,
        handlers.add_allocation_to_read_model,
    ],
    events.Deallocated: [ # 할당 해제
        handlers.remove_allocation_from_read_model,
        handlers.reallocate,
    ],
    events.OutOfStock: [handlers.send_out_of_stock_notification],
}

# src/allocation/service_layer/handler.py

#  할당시 업데이트
def add_allocation_to_read_model(
    event: events.Allocated,
    uow: unit_of_work.SqlAlchemyUnitOfWork,
):
    with uow:
        uow.session.execute(
            """
            INSERT INTO allocations_view (orderid, sku, batchref)
            VALUES (:orderid, :sku, :batchref)
            """,
            dict(orderid=event.orderid, sku=event.sku, batchref=event.batchref),
        )
        uow.commit()


# 할당 해제 시 삭제 
def remove_allocation_from_read_model(
    event: events.Deallocated,
    uow: unit_of_work.SqlAlchemyUnitOfWork,
):
    with uow:
        uow.session.execute(
            """
            DELETE FROM allocations_view
            WHERE orderid = :orderid AND sku = :sku
            """,
            dict(orderid=event.orderid, sku=event.sku),
        )
        uow.commit()

위 코드는 잘 동작할뿐 아니라 앞선 대안들과 마찬가지로 통합테스트도 잘 통과 한다

읽기 모델의 시퀀스 다이어그램 

• POST/쓰기 요청은 두 트랜잭션으로 하나는 쓰기 모델에 할당하여 커밋, 다른 하나는 읽기 모델 업데이트 
• GET/읽기 요청이 오면 이 읽기 모델을 사용해서 할당 정보를 반환해줌 

처음부터 다시 만들기

프로그램이 깨지게 된다면?
버그나 일시적인 서비스 중단으로 뷰 모델 업데이트가 일어나지 않았다면, 뷰 모델을 다시 만들면 된다 
서비스 계층을 사용해 뷰 모델을 업데이트하므로 다음과 같은 작업을 수행하는 도구를 작성할 수 있다

- 쓰기 쪽의 현재 상태를 질의해서 현재 할당된 내용을 찾는다 
- 할당된 원소마다 add_allocate_to_read_model 핸들러를 호출한다 

이렇게 데이터 이력으로부터 완전히 새로운 읽기 모델을 만들 수 있다 

읽기 모델 구현을 변경하기 쉽다

레디스를 사용해 모델을 구축하기로 한다면?

# src/allocation/service_layer/handler.py

# 레디스 읽기 모델을 업데이트하는 핸들러 
def add_allocation_to_read_model(event: events.Allocated, _):
	redis_eventpublisher.update_readmodel(event.orderid, event.sku, event.batchref)
    

def remove_allocation_from_read_model(event: events.Deallocated, _):
	redis_eventpiblisher.update_readmodel(event.orderid, event.skum None)

# src/allocation/adapters/redis_eventpublisher.py

# 레디스 모듈의 도우미 함수들은 한 줄 짜리 함수들이다
def updated_readmodel(orderid, sku, batchref):
	r.hset(orderid, sku, batchref)
    
def get_readmodel(orderid):
	return r.hgetall(orderid)

# src/allocation/views.py

# 레디스에 맞춰 변경한 뷰 
def allocations(orderid):
    batches = redis_eventpublisher.get_readmodel(orderid)
    return [
    	{"batcheref": b.decode(), "sku": s.decode()}
        for s, b in batched.items()
    ]

이미 있던 통합테스트는 세부구현에서 분리된 추상화 수준에서 작성됐기 때문에 여전히 성공한다 

마치며

다양한 뷰 모델의 트레이드 오프 

방법	장점	단점
저장소를 그냥 사용	간단하고 일관성 있는 접근가능	복잡한 패턴의 질의의 경우 성능문제 발생
ORM과 커스텀 질의 사용	DB 설정과 모델 정의 재사용가능	자체 문법이 있고 나름대로의 문제점이 있는 다른 질의 언어를 한가지 더 도입해야함
수기로 작성한 SQL 사용	표준 질의 문법을 사용해 성능을 세밀하게 제어 가능	DB 스키마 변경 시 수기로 작성한 질의와 ORM을 함께 바꿔야함. 정규화가 잘된 스키마는 여전히 성능상 한계가 있을 수 있음.
이벤트를 사용해 별도로 읽기 저장소 만들기	읽기 전용 복사본은 규모를 키우기 쉬움. 데이터가 바뀔 때 뷰를 구축해 질의를 가능한 한 간단하게 만들 수 있음	복잡한 기법, 해리는 여러분의 동기와 취향을 영원히 못미더워할 것..

(이 책의 예제에서는 할당 서비스는 단일 SKU에 대한 Batches를 기준으로 생각하지만, 사용자는 여러 SKU에 걸친 전체 주문의 할당에 신경을 쓰기 때문에 ORM을 쓰면 약간 이상해지는 것)