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是足够的。掌握了核心的基础，稍微融会贯通、举一反三其实就可以组合出大项目了。