CQRS 模式最佳化讀取操作效能

領域模型設計主要針對寫入操作，使用其進行讀取操作時，常面臨效能瓶頸。直接使用 Repository 讀取資料需要額外的迴圈和過濾，而 ORM 也可能遇到 SELECT N+1 問題。CQRS 模式提供解決方案，將讀寫操作分離，允許針對讀取操作建立專門的讀取模型。透過建立反規格化的讀取檢視或使用 Redis 等記憶體資料函式庫，可以大幅簡化查詢邏輯並提升效能。事件驅動架構進一步簡化了讀取模型的更新流程，確保資料一致性。

最佳化讀取操作的挑戰與 CQRS 介紹

在軟體開發中，領域模型（Domain Model）主要設計用於處理寫入操作（write operations），然而讀取操作（read operations）的需求往往在概念上有所不同。這種差異導致了在使用領域模型進行讀取操作時出現了許多挑戰。

使用 Repository 進行讀取操作的侷限性

首先，看看使用現有的 Repository 進行讀取操作的例子。在 allocations 函式中，我們嘗試取得特定訂單的分配批次參考：

def allocations(orderid: str, uow: unit_of_work.AbstractUnitOfWork):
    with uow:
        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
        ]

內容解密：

1. uow.products.for_order(orderid=orderid)：透過 orderid 取得相關產品。
2. batches = [b for p in products for b in p.batches]：取得所有相關批次。
3. if orderid in b.orderids：過濾出特定訂單的批次。

這種方法雖然重用了現有的抽象，但顯得相當笨拙，需要在 Python 中進行額外的迴圈和過濾操作，而這些工作本可以更高效地由資料函式庫完成。

領域模型非讀取最佳化設計

領域模型的設計主要關注業務邏輯、狀態變更和事件處理等寫入操作，而讀取操作的需求則往往被忽略。這種設計導致了在使用領域模型進行讀取操作時效率低下。

提示：領域模型的複雜度越高，就越需要做出更多的結構選擇，這些選擇使得模型越來越難以用於讀取操作。

使用 ORM 的替代方案及其侷限性

另一種方法是直接使用物件關聯對映（ORM）工具來查詢資料。例如，使用 SQLAlchemy 查詢批次：

def allocations(orderid: str, uow: unit_of_work.AbstractUnitOfWork):
    with uow:
        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
        ]

內容解密：

1. uow.session.query(model.Batch)：使用 ORM 查詢 Batch 物件。
2. .join(model.OrderLine, model.Batch._allocations)：與 OrderLine 表進行連線操作。
3. .filter(model.OrderLine.orderid == orderid)：根據 orderid 進行篩選。

雖然 ORM 簡化了資料函式庫操作，但仍存在效能問題，例如 SELECT N+1 問題。此外，ORM 可能無法最佳化複雜查詢。

CQRS：命令查詢職責分離

為瞭解決上述問題，引入了命令查詢職責分離（CQRS）模式。CQRS 主張將寫入操作（命令）和讀取操作（查詢）分開處理，以最佳化效能和簡化邏輯。

CQRS 的優勢

• 最佳化讀取操作：透過建立專門的讀取模型，可以最佳化查詢效能。
• 簡化領域模型：領域模型可以專注於業務邏輯和寫入操作，而無需顧慮讀取操作的複雜度。

建立專門的讀取檢視

一種常見的做法是建立一個反規格化的讀取檢視，以最佳化查詢效能。例如，建立一個 allocations_view 表：

CREATE TABLE allocations_view (
    orderid VARCHAR(255),
    sku VARCHAR(255),
    batchref VARCHAR(255)
);

然後，可以使用簡單的 SQL 查詢取得所需的資料：

def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
    with uow:
        results = list(uow.session.execute(
            'SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid',
            dict(orderid=orderid)
        ))
        return results

內容解密：

1. uow.session.execute：執行原生 SQL 查詢。
2. 'SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid'：查詢特定訂單的分配資訊。

這種方法透過建立專門的讀取檢視，大大簡化了查詢邏輯並最佳化了效能。

最佳化讀取模型的事件驅動架構

在處理大量讀取操作的系統中，達到關係型資料函式庫的效能極限並不罕見。即使資料函式庫索引經過最佳化，複雜的查詢仍然會消耗大量CPU資源。最快的查詢方式始終是直接根據主鍵進行查詢，例如 SELECT * FROM mytable WHERE key = :value。

這種方法不僅提升了效能，還帶來了可擴充套件性。當向關係型資料函式庫寫入資料時，需要確保對相關行進行鎖定，以避免一致性問題。如果多個客戶端同時修改資料，將會出現奇怪的競爭條件。然而，在讀取資料時，並發執行的客戶端數量沒有限制。因此，唯讀儲存可以水平擴充套件。

使用事件處理器更新讀取模型

我們為 Allocated 事件新增第二個處理器：

# src/allocation/service_layer/messagebus.py
EVENT_HANDLERS = {
    events.Allocated: [
        handlers.publish_allocated_event,
        handlers.add_allocation_to_read_model
    ],
}

更新讀取模型的程式碼如下：

# src/allocation/service_layer/handlers.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()

內容解密：

1. 事件處理器註冊：在 EVENT_HANDLERS 中註冊了 Allocated 事件的處理器，除了原有的 publish_allocated_event，還新增了 add_allocation_to_read_model。
2. SQL插入操作：在 add_allocation_to_read_model 函式中，透過 uow.session.execute 執行SQL插入陳述式，將分配資訊插入 allocations_view 表格中。
3. 事務管理：使用 with uow 確保操作的原子性，並透過 uow.commit() 提交事務。

同樣地，我們需要處理 Deallocated 事件：

# src/allocation/service_layer/messagebus.py
events.Deallocated: [
    handlers.remove_allocation_from_read_model,
    handlers.reallocate
],

# src/allocation/service_layer/handlers.py
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()

內容解密：

1. 事件處理器註冊：為 Deallocated 事件註冊了 remove_allocation_from_read_model 處理器。
2. SQL刪除操作：在 remove_allocation_from_read_model 函式中，執行SQL刪除陳述式，從 allocations_view 表格中移除相關分配資訊。
3. 事務管理：同樣使用 with uow 和 uow.commit() 確保操作的原子性和一致性。

圖表說明

此圖示展示了跨兩個請求的流程：

@startuml
note
  無法自動轉換的 Plantuml 圖表
  請手動檢查和調整
@enduml

圖表內容解密：

1. 請求流程：客戶端發起POST請求更新寫入模型，並隨後發起GET請求查詢讀取模型。
2. 模型更新：寫入模型更新後，伺服器非同步更新讀取模型。
3. 查詢流程：客戶端透過GET請求查詢讀取模型，取得最新的分配資訊。

切換至Redis作為讀取模型

我們的事件驅動架構使得更換讀取模型的實作變得非常容易。例如，切換到Redis：

# src/allocation/service_layer/handlers.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_eventpublisher.update_readmodel(event.orderid, event.sku, None)

內容解密：

1. Redis更新操作：使用 redis_eventpublisher.update_readmodel 更新Redis中的讀取模型。
2. 簡化邏輯：無需直接操作SQL，簡化了邏輯處理。

Redis操作實作如下：

# src/allocation/adapters/redis_eventpublisher.py
def update_readmodel(orderid, sku, batchref):
    r.hset(orderid, sku, batchref)

def get_readmodel(orderid):
    return r.hgetall(orderid)

內容解密：

1. Redis雜湊操作：使用Redis的雜湊結構儲存訂單的分配資訊。
2. 查詢介面：提供 get_readmodel 介面查詢指定訂單的分配資訊。

檢視層也需要相應調整：

# src/allocation/views.py
def allocations(orderid):
    batches = redis_eventpublisher.get_readmodel(orderid)
    return [
        {'batchref': b.decode(), 'sku': s.decode()}
        for s, b in batches.items()
    ]

內容解密：

1. 資料轉換：將Redis傳回的位元組資料解碼為字串。
2. 結果格式化：將分配資訊格式化為字典列表傳回。

依賴注入（Dependency Injection）與載入程式（Bootstrap）

依賴注入（DI）是軟體開發中的一種設計模式，用於管理元件之間的依賴關係。在本章中，我們將探討在Python中使用依賴注入的優缺點，並介紹如何使用載入程式（bootstrap.py）來管理依賴注入和其他初始化工作。

沒有載入程式的架構

在沒有載入程式的情況下，入口點（entrypoints）需要進行大量的初始化工作，並傳遞主要的依賴項——工作單元（Unit of Work，UoW）。如圖13-1所示。

使用載入程式的架構

使用載入程式後，可以將這些初始化工作和依賴項的傳遞集中在一個地方進行管理，如圖13-2所示。

隱式依賴與顯式依賴

在Python中，依賴通常是透過隱式匯入（implicit import）來管理的。然而，在我們的示例程式碼中，我們使用了顯式依賴（explicit dependency）來管理資料函式庫依賴。這使得在測試中可以輕鬆地替換掉真實的依賴項。

顯式依賴的例子

我們的處理函式（handlers）聲明瞭對工作單元（UoW）的顯式依賴：

def allocate(cmd: commands.Allocate, uow: unit_of_work.AbstractUnitOfWork):
    # ...

這使得我們可以在服務層測試中使用假的工作單元（Fake UoW）：

uow = FakeUnitOfWork()
messagebus.handle([...], uow)

隱式依賴的問題

雖然隱式依賴在Python中很常見，但它可能會導致測試中的問題。例如，當我們需要傳送電子郵件通知時，我們通常會使用mock.patch來類別比電子郵件傳送函式。但是，這樣會導致測試程式碼與實作細節緊密耦合。

使用載入程式的好處

使用載入程式可以將依賴注入和其他初始化工作集中在一個地方進行管理，從而使程式碼更加清晰和易於維護。

載入程式的實作

我們的載入程式（bootstrap.py）將負責管理依賴注入和其他初始化工作。這包括建立工作單元、資料函式庫連線等。

程式碼範例：

# bootstrap.py
from allocation.service_layer import unit_of_work

def bootstrap():
    # 初始化工作單元
    uow = unit_of_work.SqlAlchemyUnitOfWork()
    # ...
    return uow

內容解密：

• bootstrap函式負責初始化工作單元和其他依賴項。
• unit_of_work.SqlAlchemyUnitOfWork()建立了一個新的工作單元例項。
• 載入程式將負責管理依賴注入和其他初始化工作。

未來趨勢與實務應用評估

隨著軟體系統的複雜度增加，依賴注入和管理變得越來越重要。載入程式提供了一種集中管理依賴注入的方式，使程式碼更加清晰和易於維護。在未來的軟體開發中，依賴注入和載入程式將繼續扮演重要的角色。