一、简单入门之入门
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 } }