Golang事务模型

_张晓龙_ · · 5253 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

序言

笔者在《软件设计的演变过程》一文中，将通信系统软件的DDD分层模型最终演进为五层模型，即调度层(Schedule)、事务层(Transaction DSL)、环境层(Context)、领域层(Domain)和基础设施层(Infrastructure)，我们简单回顾一下：

ddd-layer-with-dci-dsl.png

调度层：维护UE的状态模型，只包括业务的本质状态，将接收到的消息派发给事务层。
事务层：对应一个业务流程，比如UE Attach，将各个同步消息或异步消息的处理组合成一个事务，当事务失败时，进行回滚。当事务层收到调度层的消息后，委托环境层的Action进行处理。
环境层：以Action为单位，处理一条同步消息或异步消息，将Domain层的领域对象cast成合适的role，让role交互起来完成业务逻辑。
领域层：不仅包括领域对象及其之间关系的建模，还包括对象的角色role的显式建模。
基础实施层：为其他层提供通用的技术能力，比如消息通信机制、对象持久化机制和通用的算法等

本文将聚焦于事务层，主要讨论事务模型，代码抽象层次和业务流程图一一对应。

同步模型

毫无疑问，异步模型是复杂的。但在管理域的组件中，对实时性和性能并没有极致的要求，同时协程（比如，Goroutine）非常轻量级，所以使用同步模型是一种非常聪明且简单的处理方式，如下图所示：

synchronous-model.png

在一个同步模型里，一个系统一旦发出一个请求消息，并需要等待其应答，则当前协程就会进入休眠态，直到应答消息来临或超时为止。协程可以看做是用户态轻量级的线程，占用资源非常少，当前系统同时可以有成百上千个协程运行。

假定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序列组成。

transaction.png

从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为粒度进行复用。

有疑问加站长微信联系（非本文作者）

本文来自：简书

感谢作者：_张晓龙_

查看原文：Golang事务模型

入群交流（和以上内容无关）：加入Go大咖交流群，或添加微信：liuxiaoyan-s 备注：入群；或加QQ群：692541889

5253 次点击

加入收藏微博

收入我的专栏

上一篇：golang在编译时用ldflags设置变量的值

下一篇：Go io包

context

channel

代码

业务流程

0 回复

添加一条新回复（您需要登录后才能回复没有账号？）

请尽量让自己的回复能够对别人有帮助
支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
支持 @ 本站用户；支持表情（输入 : 提示），见 Emoji cheat sheet
图片支持拖拽、截图粘贴等方式上传

关注我

扫码关注领全套学习资料
加入 QQ 群：
- 192706294（已满）
- 731990104（已满）
- 798786647（已满）
- 729884609（已满）
- 977810755（已满）
- 815126783（已满）
- 812540095（已满）
- 1006366459（已满）
- 692541889
加入微信群：liuxiaoyan-s，备注入群
也欢迎加入知识星球 Go粉丝们（免费）

Golang事务模型

序言

同步模型

事务

事务过程控制

基础数据结构

TransInfo

S1Obj

Fragment

Action

Procedure

repeat

optional

默认

事务回滚

Action

Procedure

repeat

optional

事务并发

读channel

写channel

错误和异常处理

小结

用户登录

今日阅读排行

一周阅读排行

关注我

给该专栏投稿 写篇新文章

收入到我管理的专栏 新建专栏

序言

同步模型

事务

事务过程控制

基础数据结构

TransInfo

S1Obj

Fragment

Action

Procedure

repeat

optional

默认

事务回滚

Action

Procedure

repeat

optional

事务并发

读channel

写channel

错误和异常处理

小结

给该专栏投稿写篇新文章

收入到我管理的专栏新建专栏