今天学习一下用 Go 实现观察者模式，观察者模式主要是用来实现事件驱动编程。事件驱动编程的应用还是挺广的，除了我们都知道的能够用来解耦：用户修改密码后，给用户发短信进行风险提示之类的典型场景，在微服务架构实现最终一致性、实现事件源（A ES）这些都会用到。
我们先来简单学习一下用 Go 实现观察者设计模式，给怎么实现事件驱动编程、事件源这些模式做个铺垫。主要也是我也老没看设计模式了，一起再复习一下。以前看的设计模式教程都是 Java 的，这次用 Go 实现一番。
观察者模式
咱们先来看一下观察者模式的概念，我尽量加一些自己的理解，让它变成咱们都能理解的大俗话：
概念
观察者模式 (Observer Pattern)，定义对象间的一种一对多依赖关系，使得每当一个对象状态发生改变时，其相关依赖对象皆得到通知，依赖对象在收到通知后，可自行调用自身的处理程序，实现想要干的事情，比如更新自己的状态。
发布者对观察者唯一了解的是它实现了某个接口（观察者接口）。这种松散耦合的设计最大限度地减少了对象之间的相互依赖，因此使我们能够构建灵活的系统。
我的理解
上面这段话看完，相信几乎对于理解观察者模式能起到的作用微乎其微，类似于现实职场里加班对项目进度起到的作用一样，加班的时候谁还没打过几把王者荣耀，嘿。下面我用自己的理解再给你们唠一下。
观察者模式也经常被叫做发布 - 订阅（Publish/Subscribe）模式、上面说的定义对象间的一种一对多依赖关系，一 - 指的是发布变更的主体对象，多 - 指的是订阅变更通知的订阅者对象。
发布的状态变更信息会被包装到一个对象里，这个对象被称为事件，事件一般用英语过去式的语态来命名，比如用户注册时，用户模块在用户创建好后发布一个事件 UserCreated 或者 UserWasCreated 都行，这样从名字上就能看出，这是一个已经发生过的事件。
事件发布给订阅者的过程，其实就是遍历一下已经注册的事件订阅者，逐个去调用订阅者实现的观察者接口方法，比如叫 handleEvent 之类的方法，这个方法的参数一般就是当前的事件对象。
至于很多人会好奇的，事件的处理是不是异步的？主要看我们的需求是什么，一般情况下是同步的，即发布事件后，触发事件的方法会阻塞等到全部订阅者返回后再继续，当然也可以让订阅者的处理异步执行，完全看我们的需求。
大部分场景下其实是同步执行的，单体架构会在一个数据库事务里持久化因为主体状态变更，而需要更改的所有实体类。
微服务架构下常见的做法是有一个事件存储，订阅者接到事件通知后，会把事件先存到事件存储里，这两步也需要在一个事务里完成才能保证最终一致性，后面会再有其他线程把事件从事件存储里搞到消息设施里，发给其他服务，从而在微服务架构下实现各个位于不同服务的实体间的最终一致性。
所以观察者模式，从程序效率上看，大多数情况下没啥提升，更多的是达到一种程序结构上的解耦，让代码不至于那么难维护。
Go 实现观察者模式
说了这么多，我们再看下用 Go 怎么实现最简单的观察者模式：
代码语言：javascript复制package main
import "fmt"
// Subject 接口，它相当于是发布者的定义
type Subject interface {
Subscribe(observer Observer)
Notify(msg string)
}
// Observer 观察者接口
type Observer interface {
Update(msg string)
}
// Subject 实现
type SubjectImpl struct {
observers []Observer
}
// Subscribe 添加观察者（订阅者）
func (sub *SubjectImpl) Subscribe(observer Observer) {
sub.observers = append(sub.observers, observer)
}
// Notify 发布通知
func (sub *SubjectImpl) Notify(msg string) {
for _, o := range sub.observers {
o.Update(msg)
}
}
// Observer1 Observer1
type Observer1 struct{}
// Update 实现观察者接口
func (Observer1) Update(msg string) {
fmt.Printf("Observer1: %sn", msg)
}
// Observer2 Observer2
type Observer2 struct{}
// Update 实现观察者接口
func (Observer2) Update(msg string) {
fmt.Printf("Observer2: %sn", msg)
}
func main(){
sub := &SubjectImpl{}
sub.Subscribe(&Observer1{})
sub.Subscribe(&Observer2{})
sub.Notify("Hello")
}
这就是 Go 实现观察者模式的代码，实际应用的时候，一般会定义个事件总线 EventBus 或者事件分发器 Event Dispatcher，来管理事件和订阅者间的关系，以及分发事件，它们两个就是名不一样，角色定位一样。
下面看一下用 Go 怎么实现事件总线。
Go 实现事件总线
下面我们实现一个支持以下功能的事件总线
- 异步不阻塞
- 支持任意参数值
这个代码不是我自己写的，出处见代码注释首行。
代码
代码语言：javascript复制// 代码来自https://lailin.xyz/post/observer.html
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)
}
}
单测
代码语言：javascript复制package eventbus
import (
"fmt"
"testing"
"time"
)
func sub1(msg1, msg2 string) {
time.Sleep(1 * time.Microsecond)
fmt.Printf("sub1, %s %sn", msg1, msg2)
}
func sub2(msg1, msg2 string) {
fmt.Printf("sub2, %s %sn", msg1, msg2)
}
func TestAsyncEventBus_Publish(t *testing.T) {
bus := NewAsyncEventBus()
bus.Subscribe("topic:1", sub1)
bus.Subscribe("topic:1", sub2)
bus.Publish("topic:1", "test1", "test2")
bus.Publish("topic:1", "testA", "testB")
time.Sleep(1 * time.Second)
}
毫不意外这个事件总线，只是个例子，咱也不能在项目开发里使用，这篇文章咱们先搞清概念，我其实前两天关注了下，没有发现什么好用的事件分发、事件总线的三方库，好在实现起来也不难，后面我准备自己写一个能用的到时候分享给大家，最起码是在学习、练习项目里能使用的吧。
总结
今天给大家用大白话瞎唠了一下观察者模式的原理和实际怎么应用，感觉文章的精髓主要在前半部分，可能有的不你还不能理解，后面我会再通过后续文章逐一解释，其实这些都是事件驱动和事件源这些模式里的基础内容。
至于这次给出的代码，其实没啥实战意义，就是大家先了解一下。Go 里边关于事件驱动之类的内容，感觉不多，有 Spring 使用经验的可以先看看 Spring 提供的@EventListener 注解，需要订阅者异步执行可以配合 @Async 注解使用，至于我上面说的需要保证事件发布的主体和订阅者的原子化的话，则是通过@Transitional 和 @TransactionalEventListener 结合使用来实现。
今天的内容就这些，下期再见，各位读者大佬如果方便还请转发、点赞、关注，支持一波，评论区走起来，让我们一起变得更强。
