背景
这段时间,重构了一些服务的基础工具库,主要是解耦pub-sub改为异步系统[eventbus],简单调整了定时器[clock]。本来以为已经大幅简化了业务没问题了,结果5月份,其中一个服务因为广播事件,导致死锁。分析后,发现是一个非常基础的问题导致,值得捋一捋。
问题原因大致是这样:
有一个服务对象,通过RPC,对外提供多个公共服务,并可以反向推送消息给客户端。其中,
- 有多个rpc方法,被客户端调用,有的方法需要数据保护,调用了锁。
- 该服务能收到服务端的网络断开命令,会调用名为ServerNetReset方法能够随时重置网络代理这个私有属性,也调用了锁。
- 该服务对象订阅了一些事件,一旦触发就会向客户端主动调用Notify方法来推送消息。为了避免该操作时,网络代理被ServerNetReset方法清空,因此Notify调用的私有方法push也调用了锁。
结果某一次业务重构,有一个方法Foo在使用过程中某种情况下调用了Notify推送消息,同时因为Foo有对象私有数据维护,直接使用了锁。于是,就出现了死锁状态。
相信类似于这种问题的场景,还会很多,只是我们没有发现。
最终,我们改进方案是:网络重置、消息推送这两种操作,通过消息方式串行执行,不再用锁。
使用场景
学习go的时候,很多资料都提到:“多用通道(chan),少用锁”。对于长期习惯同步编程,方法之间直接调用,对其中的理解并不深入,很多人更多把chan作为信号传递。因为异步调用涉及到事件定义、订阅发布系统、延时返回,远远没有直接调用方法来的简便。因此,一个上万行代码的项目,会使用大量的锁来保护对象属性。
如果要采用通道,不用锁,就不得不在“开发效率”、“运行效率”、“资源占用”这三个方面权衡。简单来看:
基本工具库对象,单向引用,建议用锁。
通过锁进行对象内部属性的保护,同步直接调用对象提供的公共方法,是运行效率最高的。如果该对象运行示例不需要与其他实例“相互关联”,而仅仅是被引用,则用锁是完全没问题,也是最简单的。
比如,我们做一个支持并发的计数器,不存在对第三方对象的引用,这时候,只需要用锁即可。比如:
//Counter counter is a multi-thread safe counters
type Counter struct {
mut sync.Mutex
currNum int64 //当前数量
maxNum int64 //最大数量
}
//AddOne 在原内部计数基础上,+1。
func (c *Counter) AddOne() int {
new := atomic.AddInt64(&c.currNum, 1)
c.mut.Lock()
if c.maxNum < new {
c.maxNum = new
}
c.mut.Unlock()
return int(c.currNum)
}
//DecOne 在原内部计数基础上,-1。
func (c *Counter) DecOne() int {
return int(atomic.AddInt64(&c.currNum, -1))
}
//Current 获取当前内部计数结果。
func (c *Counter) Current() int {
return int(atomic.LoadInt64(&c.currNum))
}
//MaxNum 计数器生存周期内,最大的计数。
func (c *Counter) MaxNum() int {
return int(atomic.LoadInt64(&c.maxNum))
}
//NewCounter counter constructor
func NewCounter() *Counter {
return &Counter{}
}
业务对象,尤其是DDD中提到的聚集之间,优先考虑用消息框架解耦。
由于对象间引用非常复杂,最容易理解的就是魔兽世界的战斗场景:双方多个玩家相互配合,不断施展攻击、辅助技能,过程中有的英雄使用了道具,有的被攻击导致死亡等。如果采用同步调用,对象A调用对象B,B调用C,C执行完成后,某条件下需要再告知A,那就非常复杂。这时候,我们考虑到的是ECS框架,基本的就是pub-sub系统支持。
对于pubsub的使用,不同系统接口略有不同,大家也比较熟悉,这里就不举例。
在复杂的业务对象建议用异步消息
如果一个实例存在多个公共方法+私有方法,类似于前面问题背景描述的那样,既有外部UI带来的命令驱动,又有内部的消息框架驱动。考虑到并发,不得不引入锁的时候,则建议采用串行异步方式。所有业务方法不对外,对象只有创建、接受消息、销毁三个对外的公共方法。所有消息只有一个入口,这样,就可以不用锁了。代码结构非常简单:
type Message1 struct {
}
type Message2 struct {
}
type A struct {
close int32 //对象是否关闭的标志
msgbuf chan interface{} //消息缓冲
}
func NewA() *A {
a := &A{
msgbuf: make(chan interface{}, 10),
}
go a.receive()
return a
}
func (a *A) Post(message interface{}) {
if atomic.LoadInt32(&a.close) == 1 {
a.msgbuf <- message
}
}
func (a *A) receive() {
//通过defer实现简单的故障隔离
defer func() {
if err := recover(); err != nil {
log.Println(err)
}
}()
//执行消息处理
for message := range a.msgbuf {
switch msg := message.(type) {
case Message1:
a.foo1(msg)
case Message2:
a.foo2(msg)
}
}
}
func (a *A) foo1(message Message1) {
}
func (a *A) foo2(message Message2) {
}
func (a *A) Close() {
if atomic.CompareAndSwapInt32(&a.close, 0, 1) {
// do other thing
}
}
//...
特别强调,即使作为Consumer,在pub-sub系统中订阅事件,也只传递Post作为事件响应的函数句柄,这样即使复杂系统也不会出现因为多方法执行操作内部属性,需要加锁保护,而带来的负面问题。
当然,异步消息对象的使用需要基本功,并且有额外的工作:
- 清楚架构。比如rpc服务的对象是一个,但其中被调用的方法是并行。原因是 rpc/server.go Line481 用了协程:
go service.call(server, sending, mtype, req, argv, replyv, codec)
- 前期构架不如同步调用直观,构建缓慢,效益要在维护时才能体现。
- 有的消息回复会很麻烦,需要Future、Promise、Callback这些模式,而go标准库没有原生支持。
对于Future、Promise、Callback模式,其实很简单,本质上就是异步框架提供的,用于业务对象间的值对象。go原生的waitgroup、cond、chan已经提供了很好的原生支持,自己很容易扩展。
此外,从理论到实践长达40年的actor模型也非常不错,相对于scala、java等,go的内存优势非常明显,也非常不错。因为这两部分涉及到更多设计模式的使用,内容不少,有时间另开帖说说。
有疑问加站长微信联系(非本文作者)