CQRS簡單入門(Golang)

一、簡單入門之入門

　　CQRS/ES和領域驅動設計更搭，故整體分層沿用經典的DDD四層。其實要實現的功能概要很簡單，如下圖。

 

　　基礎框架選擇了https://github.com/looplab/eventhorizon，該框架功能強大、示例都挺復雜的，囊括的概念太多，不太適合入門，所以決定在其基礎上，進行簡化。

        

二、簡化使用eventhorizon

　　Eventhorizon已經提供了詳盡的使用案例（https://github.com/looplab/eventhorizon/tree/master/examples），只是概念比較多，也不符合之前的一般使用方式，故按照概要圖進行簡化使用。

1.presentation

　　使用github.com/gin-gonic/gin，路由功能等，與業務無關都可以委托出去，同時抽象了一個核心的函數，作為銜接presentation 和application層。

從gin上下文中讀取輸入數據，並根據約定的Command Key,轉交給application層進行相應的Command解析。

func handles(command string) gin.HandlerFunc {
    return func(c *gin.Context) {
        data, err := c.GetRawData()
        if err != nil {
            c.JSON(http.StatusBadRequest, "")
            return
        }
        result := application.HandCommand(data, command)
        c.JSON(http.StatusOK, result)
    }
}

2. application

　　Application很薄的一層，依然是與業務無關的，重點在於將計算機領域的數據、模型，轉換為業務領域建模所需。

　　核心函數依然只有一個，主要功能為：創建正確的Command；將presentation層傳遞上來數據轉為為領域層所需要的模型（Command來承載）；委托“命令總線”發布命令，不必關心命令的接收方會怎樣，解除對命令執行方的依賴，只關心命令是否正確發送出去；向presentation層報告命令發布情況。

//api2Cmd  路由到領域CMD的映射
var api2Cmd map[string]eh.CommandType

type Result struct {
    Succ bool        `json:"success"`
    Code int         `json:"code"`
    Msg  string      `json:"msg"`  // message
    Data interface{} `json:"data"` // data object
}

func HandCommand(postBody []byte, commandKey string) (result Result) {
    cmd, err := eh.CreateCommand(eh.CommandType(commandKey))
    if err != nil {
        result.Msg = "could not create command: " + err.Error()
        return
    }
    if err := json.Unmarshal(postBody, &cmd); err != nil {
        result.Msg = "could not decode Json" + err.Error()
        return
    }
    ctx := context.Background()
    if err := bus.HandleCommand(ctx, cmd); err != nil {
        result.Msg = "could not handle command: " + err.Error()
        return
    }

    result.Succ = true
    result.Msg = "ok"

    return
}

3. domain

　　Domain層，核心的業務邏輯層，不進行累贅的表述，重點需要介紹下domain/Bus。總線也可以放置到infrastructure層，不過根據個人習慣寫在了domain層里。

　　Domain/Bus，整個CQRS的核心、負責命令、事件的發布、注冊等功能。核心功能主要有：命令的注冊、命令的執行、事件的注冊、事件的發布（異步）和存儲、EventStore的構建等。核心功能和方法如下：

//commandBus 命令總線
var commandBus = bus.NewCommandHandler()

//eventBus 事件總線
var eventBus = eventbus.NewEventBus(nil)

//
var eventStore eh.EventStore

//aggregateStore 領域事件存儲與發布
//var AggregateStore *events.AggregateStore

func InitBus() {
    eventStore, _ = eventstore.NewEventStore("127.0.0.1:27017", "EventStore")
    //AggregateStore, _ = events.NewAggregateStore(eventStore, eventBus)
}

//RegisterHandler 注冊命令的處理
func RegisterHandler(cmd eh.CommandType, cmdHandler eh.Aggregate) {
    err := commandBus.SetHandler(cmdHandler, cmd)
    if err != nil {
        panic(err)
    }
}

//HandleCommand 命令的執行
func HandleCommand(ctx context.Context, cmd eh.Command) error {
    return commandBus.HandleCommand(ctx, cmd)
}

//RegisterEventHandler 注冊事件的處理
func RegisterEventHandler(evtMatcher eh.EventMatcher, evtHandler eh.EventHandler) {
    eventBus.AddHandler(evtMatcher, evtHandler)
}

//RaiseEvents 異步進行事件的存儲 和 發布
func RaiseEvents(ctx context.Context, events []eh.Event, originalVersion int) error {
    go eventStore.Save(ctx, events, originalVersion)
    for _, event := range events {
        err := eventBus.PublishEvent(ctx, event)
        if err != nil {
            return err
        }
    }

    return nil
}

4. infrastructure

　　由於是簡單入門infrastructure層進行了抽象簡化，提供基本的倉儲功能。領域層進行業務處理根據所需進行數據的持久化以及讀取等。

 

三、簡要總結

　　一種方法是使其足夠簡單以至於不存在明顯的缺陷，另外一種是使其足夠復雜以至於看不出有什么問題。

　　以上組合已經能夠支持最基礎的CQRS/ES的概念和功能了。

　　現在看看如何利用已有的東西，對具體業務進行融合。

四、小試牛刀

　　支付項目中第三方支付公司需要和客戶進行簽約。需要調用支付公司的接口將用戶提交的基本信息提交給支付公司，支付公司發送驗證碼並告知用戶須知，簽約成功之后需要將協約基本信息進行保存，以后使用該協約進行代收付等資金業務。

　　單純演示，將概要設計簡化如下：獲取客戶端提交的用戶信息，校驗數據，調用第三方支付的接口，持久化到SQLite數據庫，激活領域事件存儲到MongoDB，領域事件的處理。

1. presentation

　　這里偷懶，沒有進行API路由和命令的映射，統一使用了"/api/sign_protocol"。核心代碼注冊API。

    signProtocolAPI := "/api/sign_protocol"
    router.POST(signProtocolAPI, handles(signProtocolAPI))

2. application

         Application層不需要額外代碼

3. domain

         domain層只需要commands.go、protocol.go；代碼也很簡單，command主要兩個功能承載入參、承接應用層到聚合根。

func init() {
    eh.RegisterCommand(func() eh.Command { return &SignProtocol{} })
}

const (
    SignCommand eh.CommandType = "/api/sign_protocol"
)

type SignProtocol struct {
    ID uuid.UUID
    //通道號
    AisleType string `json:"AisleType"`
    //銀行code，各平台不一樣
    BankCode string `json:"BankCode"`
    //賬戶類型
    AccountType string `json:"AccountType"`
    //賬戶屬性
    AccountProp string `json:"AccountProp"`
    //銀行卡號
    BankCardNo string `json:"BankCardNo"`
    //預留手機號
    ReservePhone string `json:"Tel"`
    //銀行卡預留的證件類型
    IDCardType string `json:"IDType"`
    //銀行卡開戶姓名
    CardName string `json:"CardName"`
    //銀行卡預留的證件號碼
    IDCardNo string `json:"IDCardNo"`
    //提示標識
    Merrem string `json:"Merrem"`
    //備注
    Remark string `json:"Remark"`
}

func (c SignProtocol) AggregateID() uuid.UUID          { return c.ID }
func (c SignProtocol) CommandType() eh.CommandType     { return SignCommand }
func (c SignProtocol) AggregateType() eh.AggregateType { return "" } //Command需要知道具體Aggregate的存在，貌似不甚合理呀！！！

         protocol.go聚合根，主要的業務邏輯。這里也很簡單，進行領域服務請求、並且進行持久化。

func init() {
    prdctAgg := &ProtocolAggregate{
        AggregateBase: events.NewAggregateBase("ProtocolAggregate", uuid.New()),
    }
    bus.RegisterHandler(SignCommand, prdctAgg)
}

type ProtocolAggregate struct {
    *events.AggregateBase
}

var _ = eh.Aggregate(&ProtocolAggregate{})

func (a *ProtocolAggregate) HandleCommand(ctx context.Context, cmd eh.Command) error {
    switch cmd := cmd.(type) {
    case *SignProtocol:
        //命令只需要確定輸入參數滿足業務校驗即可
        err := a.CheckSign()
        if err != nil {
            return err
        }
        //實際的業務可以異步進行處理
        go a.CfrmSign(cmd)

        return nil
    }
    return fmt.Errorf("couldn't handle command")
}

func (a *ProtocolAggregate) CheckSign() error {
    //校驗輸入參數是否有效，
    return nil
}

func (a *ProtocolAggregate) CfrmSign(cmd *SignProtocol) error {

    protocolOrm := &interfaces.ProtocolOrm{
        ProtocolNo:   uuid.New().String(),
        AisleType:    cmd.AisleType,
        BankCode:     cmd.BankCode,
        BankCardNo:   cmd.BankCardNo,
        ReservePhone: cmd.ReservePhone,
        CardName:     cmd.CardName,
        IDCardNo:     cmd.IDCardNo,
        Merrem:       cmd.Merrem,
        Remark:       cmd.Remark,
        Status:       1,
    }
    protocolOrm.AccountType, _ = strconv.Atoi(cmd.AccountType)
    protocolOrm.AccountProp, _ = strconv.Atoi(cmd.AccountProp)
    protocolOrm.IDCardType, _ = strconv.Atoi(cmd.IDCardType)

    //這里本應該的業務邏輯請求，通過領域服務
    //result := domainser.Allinpay.SignGetCode(protocolOrm)

    protocolRepo := infrastructure.RepoFac.ProtocolRepo
    _, err := protocolRepo.Add(protocolOrm)

    if err != nil {
        return err
    }
    ctx := context.Background()
    //業務處理成功后，激活領域事件
    bus.RaiseEvents(ctx, a.Events(), 0)

    return nil
}

4. infrastructure

         Infrastructure一般的持久化。

五、一句啰嗦

　　麻雀雖小五臟俱全，很小的一golang項目（https://github.com/KendoCross/kendocqrs），入門CQRS是足夠的。掌握了核心的基礎，稍微融會貫通、舉一反三其實就可以組合出大項目了。