背景：
公司目前有个业务，收到数据后，要分发给所有的客户端或者是业务模块，类似消息通知这样的需求，自然而然就想到了事件，观察者比较简单就自己实现以下，确保最小功能使用支持即可，其他的后期进行支持就行。
- 创建事件结构体，用来发送事件信息
// Event
// @Description: 事件信息，作用：发生的动作或事情的描述
type Event struct {
//默认false，进行同步处理；true异步处理
AsyncHandle bool
//事件名称
EventName string
//目标数据
Data any
}- 创建事件监听者
// EventListener 定义监听器；事件监听器是一个函数，它接收事件并对其作出响应
type EventListener func(*Event)- 创建分发器，这快也可以不用这些，可以写到分发管理器里也是可以的，我这边主要是为了后期方便扩展使用的
// dispatcher
// @Description: 事件分发器
type dispatcher struct {
//存储事件监听器，通过名称进行分组
listeners map[string][]EventListener
}
// NewDispatcher
//
// @Author zhaosy
// @Description: 新建分发器，不允许对外开放
// @date 2024-08-07 17:12:36
func newDispatcher() *dispatcher {
return &dispatcher{
listeners: make(map[string][]EventListener),
}
}- 创建分发管理以及相关业务
// 定义全局分发管理器
var eventDispatcherManagerObj = &eventDispatcherManager{
dispatcher: newDispatcher(),
RegisterChannel: make(chan *eventListenerInfo),
//容量给1000，后续可以根据情况进行设置大小即可
EventChannel: make(chan *Event, 1000),
}
func init() {
//异步进行启动
go eventDispatcherManagerObj.Start()
}
// eventListenerInfo
// @Description: 监听者封装，供内部使用
type eventListenerInfo struct {
EventName string
EventListener
}
// eventDispatcherManager
// @Description: 事件分发处理器，供内部使用
type eventDispatcherManager struct {
*dispatcher
RegisterChannel chan *eventListenerInfo
EventChannel chan *Event
}
// Start
//
// @Author zhaosy
// @Description: 开始启动分发处理器
// @date 2024-08-08 09:32:58
func (e *eventDispatcherManager) Start() {
for {
select {
//发送事件
case event := <-e.EventChannel:
{
//这里可以进行扩展，例如取消某个事件针对某个监听者分发
fmt.Println("监听事件", event.EventName)
//这里匹配是通过精确匹配，后期如果需要进行模糊匹配可以进行支持即可，例如前缀后缀这类的，进行扩展即可
for _, listener := range e.listeners[event.EventName] {
if event.AsyncHandle {
//如果采用异步发布事件，事件顺序无法保证，也就是乱序，这里可以根据实际标志是否进行异步分发
go listener(event) //通过协程进行处理
} else {
//默认采用同步方式进行分发事件
listener(event)
}
}
}
//注册事件
case register := <-e.RegisterChannel:
{
fmt.Println("注册事件", register.EventName)
//进行注册
e.dispatcher.listeners[register.EventName] = append(e.dispatcher.listeners[register.EventName], register.EventListener)
fmt.Printf("注册事件结果：%#v", e.dispatcher.listeners)
}
//可以扩展取消事件
}
}
}
- 监听者注册器，通过包名直接注册
// RegisterListener
//
// @Author zhaosy
// @Description: 注册事件
// @date 2024-08-08 09:05:10
func RegisterListener(eventName string, listener EventListener) error {
if eventName == "" {
return fmt.Errorf("event name is empty")
}
if listener == nil {
return fmt.Errorf("listener is nil")
}
e := &eventListenerInfo{
EventName: eventName,
EventListener: listener,
}
//发送到注册链
eventDispatcherManagerObj.RegisterChannel <- e
return nil
}- 发送监听
// Send
//
// @Author zhaosy
// @Description: 发生事件
// @date 2024-08-08 09:05:29
func Send(event *Event) error {
if event == nil {
return fmt.Errorf("event is nil")
}
if event.EventName == "" {
return fmt.Errorf("event name is empty")
}
eventDispatcherManagerObj.EventChannel <- event
return nil
}测试：
func TestEvent(t *testing.T) {
eventName := "test"
events.RegisterListener(eventName, func(event *events.Event) {
//这里建议使用goroutine进行异步处理业务，这样不会拖慢事件分发器分发效率
// go dosomething(event)
fmt.Println("第一个监听器", event.EventName, event.Data)
})
events.RegisterListener(eventName, func(event *events.Event) {
fmt.Println("第二个监听器", event.EventName, event.Data)
})
//模拟发送事件消息
for i := 0; i < 10; i++ {
//走同步
if i < 6 {
events.Send(&events.Event{
EventName: eventName,
Data: i,
})
} else {
//走异步
events.Send(&events.Event{
AsyncHandle: true,
EventName: eventName,
Data: i,
})
}
}
time.Sleep(5 * time.Second)
}结果：
