简介
在一个涉及多模块交互的系统中,如果模块的交互需要手动去调用对方的方法,那么代码的耦合度就太高了。所以产生了异步消息通信。实际上,各种各样的消息队列都是基于异步消息的。不过它们大部分都有着非常复杂的设计,很多被设计成一个独立的软件来使用。今天我们介绍一个非常小巧的异步消息通信库[message-bus]
(https://github.com/vardius/message-bus),它只能在一个进程中使用。源代码只有一个文件,我们也简单看一下实现。
快速使用
安装:
$ go get github.com/vardius/message-bus
使用:
package main
import (
"fmt"
"sync"
messagebus "github.com/vardius/message-bus"
)
func main() {
queueSize := 100
bus := messagebus.New(queueSize)
var wg sync.WaitGroup
wg.Add(2)
_ = bus.Subscribe("topic", func(v bool) {
defer wg.Done()
fmt.Println(v)
})
_ = bus.Subscribe("topic", func(v bool) {
defer wg.Done()
fmt.Println(v)
})
bus.Publish("topic", true)
wg.Wait()
}
这是官网提供的例子,message-bus
承担了模块间消息分发的角色。模块 A 和 模块 B 先向message-bus
订阅主题(topic),即告诉message-bus
对什么样的消息感兴趣。其他模块 C 产生某个主题的消息,通知message-bus
,由message-bus
分发到对此感兴趣的模块。这样就实现了模块之间的解耦,模块 A、B 和 C 之间不需要知道彼此。
上面的例子中:
- 首先,调用
messagebuss.New()
创建一个消息管理器; - 其次调用
Subscribe()
方法向管理器订阅主题; - 调用
Publish()
向管理器发布主题消息,这样订阅该主题的模块就会收到通知。
更复杂的例子
其实很多人会对何时使用这个库产生疑问,message-bus
GitHub 仓库中 issue 中至今还躺着这个问题,https://github.com/vardius/message-bus/issues/4。我是做游戏后端开发的,在一个游戏中会涉及各种各样的功能模块,它们需要了解其他模块产生的事件。例如每日任务有玩家升多少级的任务、成就系统有等级的成就、其他系统还可能根据玩家等级执行其他操作...如果硬写的话,最后可能是这样:
func (p *Player) LevelUp() {
// ...
p.DailyMission.OnPlayerLevelUp(oldLevel, newLevel)
p.Achievement.OnPlayerLevelUp(oldLevel, newLevel)
p.OtherSystem.OnPlayerLevelUp(oldLevel, newLevel)
}
需求一直在新增和迭代,如果新增一个模块,也需要在玩家升级时进行一些处理,除了实现模块自身的OnPlayerLevelUp
方法,还必须在玩家的LevelUp()
方法调用。这样玩家模块必须清楚地知道其他模块的情况。如果功能模块再多一点,而且由不同的人开发的,那么情况会更复杂。使用异步消息可有效解决这个问题:在升级时我们只需要向消息管理器发布这个升级“消息”,由消息管理器通知订阅该消息的模块。
我们设计的目录结构如下:
game
├── achievement.go
├── daily_mission.go
├── main.go
├── manager.go
└── player.go
其中manager.go
负责message-bus
的创建:
package main
import (
messagebus "github.com/vardius/message-bus"
)
var bus = messagebus.New(10)
player.go
对应玩家结构(为了简便起见,很多字段省略了):
package main
type Player struct {
level uint32
}
func NewPlayer() *Player {
return &Player{}
}
func (p *Player) LevelUp() {
oldLevel := p.level
newLevel := p.level+1
p.level++
bus.Publish("UserLevelUp", oldLevel, newLevel)
}
achievement.go
和daily_mission.go
分别是成就和每日任务(也是省略了很多无关细节):
// achievement.go
package main
import "fmt"
type Achievement struct {
// ...
}
func NewAchievement() *Achievement {
a := &Achievement{}
bus.Subscribe("UserLevelUp", a.OnUserLevelUp)
return a
}
func (a *Achievement) OnUserLevelUp(oldLevel, newLevel uint32) {
fmt.Printf("daily mission old level:%d new level:%d\n", oldLevel, newLevel)
}
// daily_mission.go
package main
import "fmt"
type DailyMission struct {
// ...
}
func NewDailyMission() *DailyMission {
d := &DailyMission{}
bus.Subscribe("UserLevelUp", d.OnUserLevelUp)
return d
}
func (d *DailyMission) OnUserLevelUp(oldLevel, newLevel uint32) {
fmt.Printf("daily mission old level:%d new level:%d\n", oldLevel, newLevel)
}
在创建这两个功能的对象时,我们订阅了UserLevelUp
主题。玩家在升级时会发布这个主题。
最后main.go
驱动整个程序:
package main
import "time"
func main() {
p := NewPlayer()
NewDailyMission()
NewAchievement()
p.LevelUp()
p.LevelUp()
p.LevelUp()
time.Sleep(1000)
}
注意,由于message-bus
是异步通信,为了能看到结果我特意加了time.Sleep
,实际开发中不太可能使用Sleep
。
最后我们运行整个程序:
$ go run .
因为要运行的是一个多文件程序,不能使用go run main.go
!
实际上,当年我因为苦于模块之间调来调去太麻烦了,自己用 C++ 撸了一个event-manager
,https://github.com/go-quiz/event-manager。思路是一样的。
缺点
message-bus
订阅主题时传入一个函数,函数的参数可任意设置,发布时必须使用相同数量的参数,这个限制感觉有点勉强。如果我们传入的参数个数不一致,程序就panic
了。我认为可以只用一个参数interface{}
,传入对象即可。例如,上面的升级事件可以使用EventUserLevelUp
的对象:
type EventUserLevelUp struct {
oldLevel uint32
newLevel uint32
}
对应地修改一下Player
的LevelUp
方法:
func (p *Player) LevelUp() {
event := &EventUserLevelUp {
oldLevel: p.level,
newLevel: p.level+1,
}
p.level++
bus.Publish("UserLevelUp", event)
}
和处理方法:
func (d *DailyMission) OnUserLevelUp(arg interface{}) {
event := arg.(*EventUserLevelUp)
fmt.Printf("daily mission old level:%d new level:%d\n", event.oldLevel, event.newLevel)
}
这样一来,我们似乎用不上反射了,订阅者都是func (interface{})
类型的函数或方法。感兴趣的可自己实现一下,我 fork 了message-bus
,做了这个修改。改动在这里:https://github.com/go-quiz/message-bus,message-bus
有测试和性能用例,改完跑一下