序言
笔者在《软件设计的演变过程》一文中,将通信系统软件的DDD分层模型最终演进为五层模型,即调度层(Schedule)、事务层(Transaction DSL)、环境层(Context)、领域层(Domain)和基础设施层(Infrastructure),我们简单回顾一下:
- 调度层:维护UE的状态模型,只包括业务的本质状态,将接收到的消息派发给事务层。
- 事务层:对应一个业务流程,比如UE Attach,将各个同步消息或异步消息的处理组合成一个事务,当事务失败时,进行回滚。当事务层收到调度层的消息后,委托环境层的Action进行处理。
- 环境层:以Action为单位,处理一条同步消息或异步消息,将Domain层的领域对象cast成合适的role,让role交互起来完成业务逻辑。
- 领域层:不仅包括领域对象及其之间关系的建模,还包括对象的角色role的显式建模。
- 基础实施层:为其他层提供通用的技术能力,比如消息通信机制、对象持久化机制和通用的算法等
本文将聚焦于事务层,主要讨论事务模型,代码抽象层次和业务流程图一一对应。
同步模型
毫无疑问,异步模型是复杂的。但在管理域的组件中,对实时性和性能并没有极致的要求,同时协程(比如,Goroutine)非常轻量级,所以使用同步模型是一种非常聪明且简单的处理方式,如下图所示:
在一个同步模型里,一个系统一旦发出一个请求消息,并需要等待其应答,则当前协程就会进入休眠态,直到应答消息来临或超时为止。协程可以看做是用户态轻量级的线程,占用资源非常少,当前系统同时可以有成百上千个协程运行。
假定Action是一条同步消息的交互,那么业务的流程图就对应一个Action序列。
事务
事务(Transaction,简写为Trans)一词来源于数据处理的概念,下面是Wikipedia 对事务的定义:
In computer science, transaction processing is information processing thatis divided into individual, indivisible operations, called transactions. Each transaction must succeed or fail as a complete unit; it cannot remain in an intermediate state.
一般情况下,一个单一场景的用户流程图就对应一个事务,而事务则由一个Action序列组成。
从S1到S2的一次同步请求处理过程中,站在S1的视角是一个Action,而站在S2的视角却是一个事务。
事务过程控制
基础数据结构
TransInfo
TransInfo是事务模型中一个非常重要的数据结构,用于事务执行过程中的数据传递,比如事务层注入到环境层的数据,Action之间串联的数据。
S1Obj
当前系统为S2,当收到来自S1的同步请求时,S2启动一个协程处理该请求。当该协程调用到调度层后,需要先初始化数据变量TransInfo和创建领域对象S1Obj,然后将它们注入到事务对象,最后执行事务。如果事务执行失败,则进行回滚。
func scheduleS1ReqTrans(req []byte) error {
transInfo := &context.TransInfo{Names: make([]string, 0)}
S1Obj := s1obj.CreateS1Obj(req)
s1ReqTrans := trans.NewS1ReqTrans()
err = s1ReqTrans.Exec(S1Obj, transInfo)
if err != nil {
s1ReqTrans.RollBack(S1Obj, transInfo)
}
return err
}
Fragment
从语义层次上看,一个Fragment是一个流程片段。
从代码层次上看,一个Fragment是一个interface。
type Fragment interface {
Exec(s1Obj *s1obj.S1Obj, transInfo *context.TransInfo) error
RollBack(s1Obj *s1obj.S1Obj, transInfo *context.TransInfo)
}
Action
Action是一条同步消息的交互,在环境层定义,是提供给事务层的原子操作,是一个Fragment,实现了Fragment接口,具体实现和业务紧密相关。
Procedure
Procedure是多条关系紧密的同步消息的交互,在事务层定义,是比Action更大的复用单元,是一个Fragment,实现了Fragment接口。
Procedure本身又是一个由Action或Procedre组成的序列,其中Action是叶子节点,Procedure是中间节点,所以Procedre是一棵多叉树。
一个通用的Procedure的代码定义如下:
type Procedure struct {
fragments []Fragment
}
创建一个具体的Procedure的代码如下:
func newA1Procedure() *Procedure {
a1Procedure := &Procedure{
fragments: []Fragment{
new(context.Action11),
newA2Procedure(),
new(context.Action12)}}
return a1Procedure
}
Procedure的执行很简单,直接调用事务层封装的原语for_each_fragments即可。
repeat
从语义层次来看,repeat用来修饰Action或Procedure,说明该Action或Procedure可以执行多次,并且至少执行一次。
从代码层次来看,repeat也是一个Fragment,因为其实现了该接口。
综上,repeat在本质上是对Action或Procedure的包装,同时也有Fragment的行为。
repeat的代码定义如下:
type repeat struct {
fragment Fragment
}
repeat的执行次数是动态确定的,即由上一个Action写入TransInfo。
有了repeat后,我们可以定义一个事务如下:
func NewS1Trans() *Transaction {
s1Trans := &Transaction{
fragments: []Fragment{
new(context.Action1),
new(context.Action2),
repeat{fragment:newA1Procedure()},
new(context.Action3)}}
return s1Trans
}
optional
optional与repeat类似^-^。
从语义层次来看,optional用来修饰Action或Procedure,说明该Action或Procedure最多执行一次,并且可以不执行。
从代码层次来看,optional也是一个Fragment,因为其实现了该接口。
综上,optional在本质上是对Action或Procedure的包装,同时也有Fragment的行为。
optional的执行次数由谓词Specification确定,Specification是一个interface,它的定义如下:
type Specification interface {
Ok(s1Obj *s1obj.S1Obj, transInfo *context.TransInfo) bool
}
谓词的实例来自两个方面的确认:
- 系统的某个开关是否打开,即开关打开时,谓词为真,执行一次Action或Procedure,否则执行零次。
- 系统的当前状态是否满足某个条件,即条件满足时,谓词为真,执行一次Action或Procedure,否则执行零次。
有了optional后,我们可以定义一个事务如下:
func NewS1Trans() *Transaction {
s1Trans := &Transaction{
fragments: []Fragment{
new(context.Action1),
optional{spec:new(context.ShouldExecAction2),
fragment:new(context.Action2)},
new(context.Action3),
repeat{fragment:newA1Procedure()},
new(context.Action4)}}
return s1Trans
}
默认
从语义层次来看, 没有repeat或optional修饰的Action或Procedure就是默认的情况,说明Action或Procedure仅且执行一次。
事务回滚
对于事务来说,执行要么成功,要么失败。当事务执行失败时,必须触发回滚,使得系统无资源泄露或残留。
当事务执行失败时,肯定实在某一个Fragment执行时失败,我们记作fragments[i],事务回滚的过程为:
- fragments[i]完成自己已分配的资源的回收和自己已写入的数据的清理;
- 从fragments[i-1]到fragments[0],依次调用它的RollBack方法。
Action
Action是事务的原子执行者,从叶子节点来看,事务都是Action序列。
当某个Action执行失败时,在Exec方法内进行该Action相关的资源回收或数据清理,不会调用该Action的RollBack函数。
Action的RollBack方法实现很简单,仅进行该Action相关的所有资源回收和数据清理。
举个例子:
Action5在执行失败前,打开了文件file1,在表table1中写了一条记录,那么它在返回error前要删除表table1中的记录,并关闭文件file1,即逆序的进行资源回收和数据清理。
至于Action1到Action4中打开了什么资源或写了什么数据,Action5一点都不care。
Action5返回错误后,事务回滚框架会自动依次调用[Action4,Action3, Action2, Action1]的Rollback函数,从而完成事务的回滚。
Procedure
如果Procedure执行失败,则在Exec方法中进行“错误处理”:
func (this *Procedure) Exec(knitterObj *knitterobj.KnitterObj, transInfo *context.TransInfo) error {
index, err := for_each_fragments(this.fragments, knitterObj, transInfo)
if err != nil {
if index <= 0 {
return err
}
back_each_fragments(this.fragments, knitterObj, transInfo, index)
}
return err
}
Exec方法在实现中使用了事务层的原语for_each_fragments和back_each_fragments:
- 对于for_each_fragments原语,在事务过程控制一节中已经提过,即正向依次遍历fragments,调用它的Exec方法。
- 对于back_each_fragments原语,先对入参index(最后一个参数)进行减一(index--),然后从index开始反向遍历fragments,调用它的RollBack方法。
如果Procedure执行成功,回滚时直接调用RollBack方法即可:
func (this *Procedure) RollBack(knitterObj *knitterobj.KnitterObj, transInfo *context.TransInfo) {
back_each_fragments(this.fragments, knitterObj, transInfo, len(this.fragments))
}
repeat
如果repeat执行失败,则进行“错误处理”:
func (this repeat) Exec(knitterObj *knitterobj.KnitterObj, transInfo *context.TransInfo) error {
for i := 0; i < transInfo.Times; i++ {
transInfo.RepeatIdx = i
err := this.fragment.Exec(knitterObj, transInfo)
if err != nil {
if i == 0 {
return err
}
i--
for j := i; j >= 0; j-- {
transInfo.RepeatIdx = j
this.fragment.RollBack(knitterObj, transInfo)
}
return err
}
}
return nil
}
这里的transInfo.RepeatIdx需要解释一下:
- 在this.fragment.Exec之前赋值为i,是为了在repeat执行Action或Procedure时,找到对应的领域对象。
- this.fragment.RollBack之前赋值为j,是为了repeat在“错误处理”时,即回滚已经完成的Action或Procedure时,找到对应的领域对象。举个例子,比如repeat的最大次数是5,当进行到第4次时发生了错误,这时需要回滚前三次的Action或Procedure。
如果repeat执行成功,回滚时直接调用RollBack方法即可:
func (this repeat) RollBack(knitterObj *knitterobj.KnitterObj, transInfo *context.TransInfo) {
for i := transInfo.Times; i >= 0; i-- {
transInfo.RepeatIdx = i
this.fragment.RollBack(knitterObj, transInfo)
}
}
optional
optional就比较简单了,如果执行过程中发生了错误,则啥也不用干,因为Action或Procedure已完成了错误处理,如下所示:
func (this optional) Exec(s1Obj *s1obj.S1Obj, transInfo *context.TransInfo) error {
if this.spec.Ok(s1Obj, transInfo) {
this.isExec = true
return this.fragment.Exec(s1Obj, transInfo)
}
return nil
}
如果optional执行成功,回滚时需要根据是否执行过Action或Procedure来进行Action或Procedure的回滚,如下所示:
func (this optional) RollBack(s1Obj *s1obj.S1Obj, transInfo *context.TransInfo) {
if this.isExec {
this.fragment.RollBack(s1Obj, transInfo)
}
}
事务并发
事务的执行过程是一个同步模型,而事务之间却是异步的。多个事务间可能共享资源,所以要对事务进行并发控制。
在Golang中,协程之间的并发控制一般使用channel,非常简单且高效。
假设一组协程使用一个共享资源,这时通过一个channel控制,那么多组协程就需要多个channel来控制。我们可以使用map,key为shareId,value为channel。
读channel
根据业务流程,要在某个Specification(谓词,optional的第一个参数)中读channel。假设该谓词为IsSomethingNotExist,示例代码如下:
func (this *IsSomethingNotExist) Ok(s1Obj *s1obj.S1Obj, transInfo *TransInfo) bool {
...
<- transInfo.Chan
transInfo.ChanFlag = true
...
}
要读channel,必须先注入。根据局部化原则,我们在谓词IsSomethingNotExist中进行注入,而不在前面的Action或Specification中进行注入,于是示例代码变为:
func (this *IsSomethingNotExist) Ok(s1Obj *s1obj.S1Obj, transInfo *TransInfo) bool {
...
concurrencyctrl.ChanMapLock.Lock()
value, ok := concurrencyctrl.ChanMap[shareId]
if ok {
transInfo.Chan = value
} else {
transInfo.Chan = make(chan int, 1)
transInfo.Chan <- 1
concurrencyctrl.ChanMap[shareId] = transInfo.Chan
}
concurrencyctrl.ChanMapLock.Unlock()
<- transInfo.Chan
transInfo.ChanFlag = true
...
}
写channel
根据业务流程,要在读channel的Specification之后的某个Action中写channel。假设该Action为DiscussAction,示例代码如下:
func (this *DiscussAction) Exec(s1Obj *s1obj.S1Obj, transInfo *TransInfo) error {
...
transInfo.Chan <- 1
transInfo.ChanFlag = false
...
}
细心的读者可能已经发现,上面的描述“要在读channel的Specification之后的某个Action中写channel”存在两种情况:
- 该Specification是optional的第一个参数,而该Action或包含该Action的Procedure是对应的第二个参数
- 该Action在该Specification对应的optional操作之后
不管Specification的Ok方法是否返回true,第二种情况总是会进行写channel操作,而第一种情况则未必,即当Specification的Ok方法返回为false时,并不会进行写channel操作,所以有瑕疵。该瑕疵的修复方法是在该Specification的Ok方法内进行判断,如果返回值为false,则进行写channel操作。假设该谓词为IsSomethingNeedDel,则示例代码为:
func (this *IsSomethingNeedDel) Ok(s1Obj *s1obj.S1Obj, transInfo *TransInfo) bool {
...
concurrencyctrl.ChanMapLock.Lock()
value, ok := concurrencyctrl.ChanMap[shareId]
if ok {
transInfo.Chan = value
} else {
transInfo.Chan = make(chan int, 1)
transInfo.Chan <- 1
concurrencyctrl.ChanMap[shareId] = transInfo.Chan
}
concurrencyctrl.ChanMapLock.Unlock()
<- transInfo.Chan
transInfo.ChanFlag = true
...
if flag {
log.Infof("***IsSomethingNeedDel: true***")
} else {
transInfo.Chan <- 1
transInfo.ChanFlag = false
log.Infof("***IsSomethingNeedDel: false***")
}
return flag
}
错误和异常处理
在事务执行过程中,不管是遇到错误还是发生了异常(panic),可能会出现对于channel读了没有写的情况,即在事务处理过程中没有实现channel的闭合操作,这将导致该组的其他协程(Goroutine)也阻塞了。
该问题的解决思路是在事务调度的入口方法中使用defer修饰的闭包对异常进行捕获,同时针对错误或异常都对channel尝试闭合操作,示例代码如下:
func scheduleS1ReqTrans(req []byte) (err error) {
transInfo := &context.TransInfo{Names: make([]string, 0)}
defer func() {
if p := recover(); p != nil {
str, ok := p.(string)
if ok {
err = errors.New(str)
} else {
err = errors.New("panic")
}
log.Info("S1ReqTrans panic recover start!")
log.Error("Stack:", string(debug.Stack()))
log.Info("S1ReqTrans panic recover end!")
}
if transInfo.ChanFlag {
transInfo.Chan <- 1
}
}()
S1Obj := s1obj.CreateS1Obj(req)
s1ReqTrans := trans.NewS1ReqTrans()
err = s1ReqTrans.Exec(S1Obj, transInfo)
if err != nil {
s1ReqTrans.RollBack(S1Obj, transInfo)
}
return err
}
小结
在管理域的组件中,对实时性和性能并没有极致的要求,同时Goroutine非常轻量级,所以使用同步模型是一种非常聪明且简单的处理方式。本文所讨论的事务模型针对的就是同步过程,先详细阐述了事务的过程控制,然后对事务的回滚给出了通用的设计框架,最后对事务的并发控制给出了简单高效的解决方案。事务模型在DDD的分层架构中位于第四层,代码抽象层次高且表达力强,和业务流程图一一对应,同时代码可以以Action或Procedure为粒度进行复用。
有疑问加站长微信联系(非本文作者)