当前位置：首页 > news >正文

go 事件机制（观察者设计模式）

news 2025/7/8 14:06:04

背景：

公司目前有个业务，收到数据后，要分发给所有的客户端或者是业务模块，类似消息通知这样的需求，自然而然就想到了事件，观察者比较简单就自己实现以下，确保最小功能使用支持即可，其他的后期进行支持就行。

创建事件结构体，用来发送事件信息

// 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 stringEventListener
}// eventDispatcherManager
// @Description: 事件分发处理器，供内部使用
type eventDispatcherManager struct {*dispatcherRegisterChannel chan *eventListenerInfoEventChannel 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 <- ereturn 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 <- eventreturn 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)
}

结果：