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解析。
1 func handles(command string) gin.HandlerFunc { 2 return func(c *gin.Context) { 3 data, err := c.GetRawData() 4 if err != nil { 5 c.JSON(http.StatusBadRequest, "") 6 return 7 } 8 result := application.HandCommand(data, command) 9 c.JSON(http.StatusOK, result) 10 } 11 }
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是足够的。掌握了核心的基础，稍微融会贯通、举一反三其实就可以组合出大项目了。