Golang事件總線機(jī)制的實(shí)現(xiàn)

【導(dǎo)讀】本文介紹了事件總線實(shí)現(xiàn)。

最近在學(xué)習(xí)開(kāi)源項(xiàng)目Grafana的代碼，發(fā)現(xiàn)作者實(shí)現(xiàn)了一個(gè)事件總線的機(jī)制，在項(xiàng)目里面大量應(yīng)用，效果也非常好，代碼也比較簡(jiǎn)單，介紹給大家看看。

源碼文件地址：grafana/bus.go at main · grafana/grafana · GitHub

1.注冊(cè)和調(diào)用

在這個(gè)項(xiàng)目里面隨處可見(jiàn)這種寫法：

funcValidateOrgAlert(c*models.ReqContext){
id:=c.ParamsInt64(":alertId")

query:=models.GetAlertByIdQuery{Id:id}

iferr:=bus.Dispatch(&query);err!=nil{
c.JsonApiErr(404,"Alertnotfound",nil)
return
}

ifc.OrgId!=query.Result.OrgId{
c.JsonApiErr(403,"Youarenotallowedtoedit/viewalert",nil)
return
}
}

關(guān)鍵是bus.Dispatch(&query)這段代碼，它的參數(shù)是一個(gè)結(jié)構(gòu)體GetAlertByIdQuery，內(nèi)容如下：

typeGetAlertByIdQuerystruct{
Idint64
Result*Alert
}

根據(jù)名字可以看出這個(gè)方法就是通過(guò)Id去查詢Alert，其中Alert結(jié)構(gòu)體就是結(jié)果對(duì)象，這里就不貼出來(lái)了。

通過(guò)查看源碼可以得知，Dispatch背后是調(diào)用了GetAlertById這個(gè)方法，然后把結(jié)果賦值到query參數(shù)的Result中返回。

funcGetAlertById(query*models.GetAlertByIdQuery)error{
alert:=models.Alert{}
has,err:=x.ID(query.Id).Get(&alert)
if!has{
returnfmt.Errorf("couldnotfindalert")
}
iferr!=nil{
returnerr
}
query.Result=&alert
returnnil
}

問(wèn)題來(lái)了，這是怎么實(shí)現(xiàn)的呢？Dispatch到底做了哪些操作？這樣做有什么好處？

下面我來(lái)一一解答：

首先，在Dispatch之前，你需要先注冊(cè)這個(gè)方法，也就是調(diào)用AddHandler，在這個(gè)項(xiàng)目里面可以看到init函數(shù)里面有大量這樣的代碼：

funcinit(){
bus.AddHandler("sql",SaveAlerts)
bus.AddHandler("sql",HandleAlertsQuery)
bus.AddHandler("sql",GetAlertById)
...
}

其實(shí)這個(gè)方法的邏輯也很簡(jiǎn)單，所謂注冊(cè)也就是把通過(guò)一個(gè)map把函數(shù)名和對(duì)應(yīng)的函數(shù)做一個(gè)映射關(guān)系保存起來(lái)，當(dāng)我們Dispatch的時(shí)候其實(shí)就是通過(guò)參數(shù)名查找之前注冊(cè)過(guò)的函數(shù)，然后通過(guò)反射調(diào)用該函數(shù)。

Bus結(jié)構(gòu)體里面有幾個(gè)map成員，在這個(gè)項(xiàng)目里面作者定義了3種不同類型的handler，一種是普通的handler，也就是剛才展示的那種，第二種是帶上下文的handler，還有一種則是事件訂閱用到的handler，我們給一個(gè)事件注冊(cè)多個(gè)監(jiān)聽(tīng)者，當(dāng)事件觸發(fā)的時(shí)候會(huì)依次調(diào)用多個(gè)監(jiān)聽(tīng)函數(shù)，其實(shí)就是一個(gè)觀察者模式。

//InProcBusdefinesthebusstructure
typeInProcBusstruct{
handlersmap[string]HandlerFunc
handlersWithCtxmap[string]HandlerFunc
listenersmap[string][]HandlerFunc
txMngTransactionManager
}

下面就看看具體的源碼，AddHandler方法內(nèi)容如下：

func(b*InProcBus)AddHandler(handlerHandlerFunc){
handlerType:=reflect.TypeOf(handler)
queryTypeName:=handlerType.In(0).Elem().Name()//獲取函數(shù)第一個(gè)參數(shù)的名稱，在上面例子里面就是GetAlertByIdQuery
b.handlers[queryTypeName]=handler
}

Dispatch方法的源碼如下：

func(b*InProcBus)Dispatch(msgMsg)error{
varmsgName=reflect.TypeOf(msg).Elem().Name()

withCtx:=true
handler:=b.handlersWithCtx[msgName]//根據(jù)參數(shù)名查找注冊(cè)過(guò)的函數(shù)，先查找?guī)tx的handler
ifhandler==nil{
withCtx=false
handler=b.handlers[msgName]
ifhandler==nil{
returnErrHandlerNotFound
}
}
varparams=[]reflect.Value{}
ifwithCtx{
//如果查找到的handler是帶Ctx的就給個(gè)默認(rèn)的Background的Ctx
params=append(params,reflect.ValueOf(context.Background()))
}
params=append(params,reflect.ValueOf(msg))

ret:=reflect.ValueOf(handler).Call(params)//通過(guò)反射機(jī)制調(diào)用函數(shù)
err:=ret[0].Interface()
iferr==nil{
returnnil
}
returnerr.(error)
}

對(duì)于AddHandlerCtx和DispatchCtx這個(gè)2個(gè)方法基本上是一樣的，只不過(guò)多了一個(gè)上下文參數(shù)，可以拿來(lái)做超時(shí)控制或者其它用途。

2.訂閱和發(fā)布

除此之外，還有2個(gè)方法AddEventListener和Publish，即事件的訂閱和發(fā)布。

func(b*InProcBus)AddEventListener(handlerHandlerFunc){
handlerType:=reflect.TypeOf(handler)
eventName:=handlerType.In(0).Elem().Name()
_,exists:=b.listeners[eventName]
if!exists{
b.listeners[eventName]=make([]HandlerFunc,0)
}
b.listeners[eventName]=append(b.listeners[eventName],handler)
}

查看源碼可以得知，可以給一個(gè)事件注冊(cè)多個(gè)handler函數(shù)，而Publish的時(shí)候則是依次調(diào)用注冊(cè)的函數(shù)，邏輯也不復(fù)雜。

func(b*InProcBus)Publish(msgMsg)error{
varmsgName=reflect.TypeOf(msg).Elem().Name()
varlisteners=b.listeners[msgName]

varparams=make([]reflect.Value,1)
params[0]=reflect.ValueOf(msg)

for_,listenerHandler:=rangelisteners{
ret:=reflect.ValueOf(listenerHandler).Call(params)
e:=ret[0].Interface()
ife!=nil{
err,ok:=e.(error)
ifok{
returnerr
}
returnfmt.Errorf("expectedlistenertoreturnanerror,got'%T'",e)
}
}
returnnil
}

這里面有一點(diǎn)不好，所有訂閱函數(shù)的調(diào)用是順序的，并沒(méi)有使用協(xié)程，所以如果注冊(cè)了很多個(gè)函數(shù)，這樣效率也不高啊。

3.好處

可能有人會(huì)好奇，為什么明明可以直接調(diào)用函數(shù)就行，為啥非得繞個(gè)彎子，整這么復(fù)雜？

況且，每次調(diào)用都得使用反射機(jī)制，性能也不行。

我覺(jué)得主要有以下幾點(diǎn)：

1.這種寫法邏輯清晰，解耦

2.方便單元測(cè)試

3.性能不是最大考量，雖然說(shuō)反射會(huì)降低性能