一、背景

• 最近看到zookeeper相关的文章，发现其本质是文件系统+观察者设计模式的分布式管理框架，包括 consul 等框架也是使用了观察者设计模式，而观察者模式实际上与 sync.Cond (opens new window) 库功能类似，准备写一篇相关的文章。

二、简单介绍

• 观察者模式是一种行为设计模式，允许发布者维护一组观察者，并在自身状态改变时自动通知这些观察者；
• 这种模式常用于实现发布-订阅系统，能够有效处理多个观察者的需求；
• 主要特点包括：
• 解耦：发布者与观察者之间没有直接依赖，观察者可以独立变化；
• 动态关系：观察者可以在运行时注册或注销，不影响发布者的正常运行；
• 事件驱动：适用于事件驱动的系统，如实时数据更新等；

三、设计实现

1，定义观察者接口

• 定义了 Deal(msg string) 方法，用于处理发生的事件消息；

// Observer 观察者接口
type Observer interface {
	Deal(msg string)
}

2，观察者实现

• 定义某个类型，实现 Deal(msg string) 方法， 按照其自定义的逻辑处理消息；

type Observer1 struct{}

func (Observer1) Deal(msg string) {
  // 按照其自定义的逻辑处理消息
	fmt.Printf("Observer1: %s\n", msg)
}

type Observer2 struct{}

func (Observer2) Deal(msg string) {
  // 按照其自定义的逻辑处理消息
	fmt.Printf("Observer2: %s\n", msg)
}

3，定义发布者接口

• 定义Subscribe(observer Observer)方法，用于接收订阅消息的观察者；
• 定义Notify(msg string)方法，用于通知观察者有事件发生。

// Subject 接口，它相当于是发布者的定义
type Subject interface {
	Subscribe(observer Observer)
	Notify(msg string)
}

4，发布者实现

• 定义 SubjectImpl结构体，并定义observers []Observer字段，用于接收订阅的观察者；
• 实现Subscribe(observer Observer)方法，用于接收注册时间的观察者，并保存下来；
• 实现Notify(msg string)方法，当事件发生时，遍历已经注册的事件观察者者，异步调用观察者实现的 Deal(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) {
	var w sync.WaitGroup
	w.Add(len(sub.observers))
	for _, o := range sub.observers {
		go func(ob Observer) {
			defer w.Done()
			ob.Deal(msg)
		}(o)
	}
	w.Wait() // 异步调用，等待订阅着执行完成
}

5，完整示例

package main

import (
	"fmt"
	"sync"
)

type Subject interface {
	Subscribe(observer Observer)
	Notify(msg string)
}

type Observer interface {
	Deal(msg string)
}

type SubjectImpl struct {
	observers []Observer
}

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

func (sub *SubjectImpl) Notify(msg string) {
	var w sync.WaitGroup
	w.Add(len(sub.observers))
	for _, o := range sub.observers {
		go func(ob Observer) {
			defer w.Done()
			ob.Deal(msg)
		}(o)
	}
	w.Wait() // 异步调用，等待订阅着执行完成
}

type Observer1 struct{}
func (Observer1) Deal(msg string) {
	fmt.Printf("Observer1: %s\n", msg)
}

type Observer2 struct{}
func (Observer2) Deal(msg string) {
	fmt.Printf("Observer2: %s\n", msg)
}

func main() {
	sub := &SubjectImpl{}
	sub.Subscribe(&Observer1{})
	sub.Subscribe(&Observer2{})
	sub.Notify("Hello")
}