Error message here!

Hide Error message here!

忘记密码?

Error message here!

请输入正确邮箱

Hide Error message here!

密码丢失?请输入您的电子邮件地址。您将收到一个重设密码链接。

Error message here!

返回登录

Close

CQRS简单入门(Golang)

KendoCross 2019-01-29 11:10:00 阅读数:266 评论数:0 点赞数:0 收藏数:0

一、简单入门之入门

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

 

版权声明
本文为[KendoCross]所创,转载请带上原文链接,感谢
https://www.cnblogs.com/KendoCross/p/10333145.html

编程之旅,人生之路,不止于编程,还有诗和远方。
阅代码原理,看框架知识,学企业实践;
赏诗词,读日记,踏人生之路,观世界之行;

支付宝红包,每日可领