Go语言设计模式：实现观察者模式与异步事件总线

顾言

发布时间：2026-10-16 | 5人浏览过

原创

通过Go语言的接口特性实现观察者模式，将发布者与订阅者解耦。文章详细解析了基础的通知分发原理，并提供了包含并发控制和反射参数的异步事件总线完整代码，帮助开发者构建高内聚、低耦合的事件驱动系统。

核心概念与价值

观察者模式（Observer Pattern）定义了对象之间的一对多依赖关系，确保当一个对象状态改变时，所有依赖它的对象都能收到通知并自动更新。这种松散耦合的设计允许系统在运行时动态调整行为，是构建灵活应用的基础。

在实际开发中，该模式常被称为发布-订阅模式。核心组件包括：

发布者（Subject）：负责维护观察者列表，并在状态变更时触发通知。
订阅者（Observer）：实现统一接口以接收通知并执行具体逻辑。
事件（Event）：封装状态变更信息的数据结构，通常使用过去式命名（如 UserCreated）表明动作已完成。

基础实现步骤

第1步：定义接口

Go语言没有继承概念，主要通过接口实现多态。需定义两个核心接口：

Subscribe(observer Observer)：用于添加观察者。
Notify(msg string)：用于向所有已注册观察者发送消息。
Update(msg string)：观察者必须实现的回调方法。

第2步：实现发布者逻辑

发布者内部维护一个切片存储观察者实例。发送通知时，遍历切片并逐个调用 Update 方法。默认情况下，这是同步执行的过程，发布者会等待所有观察者处理完毕后再继续。

第3步：配置订阅者

订阅者只需定义结构体并实现 Update 方法即可接入系统。不同订阅者可处理同一事件的多种业务逻辑，彼此互不干扰。

进阶：构建异步事件总线

在生产环境中，同步阻塞可能影响性能。通过引入 goroutine 和 reflect 包，可以实现支持任意参数且非阻塞的异步事件总线（AsyncEventBus）。

关键功能点

泛型参数支持：利用反射技术接收不定数量的参数，适配不同场景的事件数据。
异步执行：发布时启动新协程处理，立即返回结果，避免主流程阻塞。
线程安全：使用互斥锁管理订阅关系的读写。

代码结构说明

1. 基础观察者模型示例

package main

import "fmt"

// Subject 接口定义
 type Subject interface {
	Subscribe(observer Observer)
	Notify(msg string)
}

// Observer 接口定义
type Observer interface {
	Update(msg string)
}

// SubjectImpl 实现类
type SubjectImpl struct {
	observers []Observer
}

func (sub *SubjectImpl) Subscribe(o Observer) {
	sub.observers = append(sub.observers, o)
}

func (sub *SubjectImpl) Notify(msg string) {
	for _, obs := range sub.observers {
		obs.Update(msg)
	}
}

// 具体观察者
type Logger struct{}
func (l *Logger) Update(msg string) {
	fmt.Println("Log: ", msg)
}

func main() {
	bus := &SubjectImpl{}
	bus.Subscribe(&Logger{})
	bus.Notify("System started")
}

2. 异步事件总线核心实现

以下代码展示了如何利用反射实现灵活的事件分发：

package eventbus

import (
	"fmt"
	"reflect"
	"sync"
)

type Bus interface {
	Subscribe(topic string, handler interface{}) error
	Publish(topic string, args ...interface{})
}

type AsyncEventBus struct {
	handlers map[string][]reflect.Value
	lock     sync.Mutex
}

func NewAsyncEventBus() *AsyncEventBus {
	return &AsyncEventBus{
		handlers: make(map[string][]reflect.Value),
	}
}

// Subscribe 添加回调函数
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
	bus.lock.Lock()
	defer bus.lock.Unlock()

	v := reflect.ValueOf(f)
	if v.Kind() != reflect.Func {
		return fmt.Errorf("handler must be a function")
	}

	bus.handlers[topic] = append(bus.handlers[topic], v)
	return nil
}

// Publish 异步发布事件
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
	funcs := bus.handlers[topic]
	if len(funcs) == 0 {
		return
	}

	// 准备参数值
	params := make([]reflect.Value, len(args))
	for i, arg := range args {
		params[i] = reflect.ValueOf(arg)
	}

	// 并发执行每个监听器
	for _, fn := range funcs {
		go fn.Call(params)
	}
}