观察者模式主要是用来实现事件驱动编程。事件驱动编程的应用广泛,除了能用来解耦:用户修改密码后,给用户发短信进行风险提示之类的典型场景。
观察者模式:订阅(Publish/Subscribe)模式,定义对象间的一种一对多依赖关系,使得每当一个对象状态发生改变时,其相关依赖对象皆得到通知,依赖对象在收到通知后,可自行调用自身的处理程序,实现想要干的事情,比如更新自己的状态。发布者对观察者唯一了解的是它实现了某个接口(观察者接口)。这种松散耦合的设计最大限度地减少了对象之间的相互依赖,因此能够构建灵活的系统。
观察者模式的订阅是在事件类中添加自己的事件函数,而发布是事件改变时,事件类遍历自己存储的所有事件函数逐一执行。接下来实现一个互斥的发布订阅总线
package eventbus
import (
"fmt"
"reflect"
"sync"
)
// Bus Bus
type Bus interface {
Subscribe(topic string, handler interface{}) error
Publish(topic string, args ...interface{})
}
// AsyncEventBus 异步事件总线
type AsyncEventBus struct {
handlers map[string][]reflect.Value
lock sync.Mutex
}
// NewAsyncEventBus new
func NewAsyncEventBus() *AsyncEventBus {
return &AsyncEventBus{
handlers: map[string][]reflect.Value{},
lock: sync.Mutex{},
}
}
// Subscribe 订阅
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
bus.lock.Lock()
defer bus.lock.Unlock()
v := reflect.ValueOf(f) // 获取处理函数的反射值
if v.Type().Kind() != reflect.Func { // 检查处理函数是否为函数类型
return fmt.Errorf("handler is not a function")
}
handler, ok := bus.handlers[topic] // 获取当前主题的处理函数列表
if !ok {
handler = []reflect.Value{} // 如果列表不存在,则初始化为空切片
}
handler = append(handler, v) // 将处理函数添加到列表中
bus.handlers[topic] = handler // 更新处理函数列表
return nil
}
// Publish 发布
// 这里异步执行,并且不会等待返回结果
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
handlers, ok := bus.handlers[topic] // 获取当前主题的处理函数列表
if !ok {
fmt.Println("not found handlers in topic:", topic)
return
}
params := make([]reflect.Value, len(args)) // 创建一个与参数数量相同的反射值切片
for i, arg := range args {
params[i] = reflect.ValueOf(arg) // 将参数转换为反射值
}
for i := range handlers {
go handlers[i].Call(params) // 在新的goroutine中异步调用处理函数
}
}
参考公众号网管叨bi叨