Golang观察者模式实现实时数据更新

package main

import (
    "fmt"
    "sync"
    "time"
)

// Observer 接口定义了观察者接收更新的方法
type Observer interface {
    Update(data interface{})
    GetID() string // 用于识别观察者，方便注销
}

// Subject 接口定义了主题管理观察者和通知的方法
type Subject interface {
    Register(observer Observer)
    Deregister(observer Observer)
    Notify(data interface{})
}

// ConcreteObserver 是一个具体的观察者实现
type ConcreteObserver struct {
    ID string
}

func (o *ConcreteObserver) Update(data interface{}) {
    fmt.Printf("观察者 %s 收到更新: %v\n", o.ID, data)
}

func (o *ConcreteObserver) GetID() string {
    return o.ID
}

// DataSubject 是一个具体的主题实现
type DataSubject struct {
    observers map[string]Observer // 使用map方便查找和删除
    data      interface{}
    mu        sync.RWMutex      // 读写锁保护observers和data
}

func NewDataSubject() *DataSubject {
    return &DataSubject{
        observers: make(map[string]Observer),
    }
}

func (s *DataSubject) Register(observer Observer) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.observers[observer.GetID()] = observer
    fmt.Printf("观察者 %s 已注册。\n", observer.GetID())
}

func (s *DataSubject) Deregister(observer Observer) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if _, ok := s.observers[observer.GetID()]; ok {
        delete(s.observers, observer.GetID())
        fmt.Printf("观察者 %s 已注销。\n", observer.GetID())
    }
}

func (s *DataSubject) Notify(data interface{}) {
    s.mu.RLock() // 只读访问observers
    defer s.mu.RUnlock()
    s.data = data // 更新主题的内部数据
    fmt.Printf("主题状态更新为: %v, 开始通知所有观察者...\n", data)
    for _, observer := range s.observers {
        // 可以在这里启动goroutine异步通知，提高并发性
        go observer.Update(data)
    }
}

// SetData 模拟主题数据变化
func (s *DataSubject) SetData(data interface{}) {
    s.Notify(data)
}

func main() {
    // 创建主题
    subject := NewDataSubject()

    // 创建观察者
    observer1 := &ConcreteObserver{ID: "ObserverA"}
    observer2 := &ConcreteObserver{ID: "ObserverB"}
    observer3 := &ConcreteObserver{ID: "ObserverC"}

    // 注册观察者
    subject.Register(observer1)
    subject.Register(observer2)
    subject.Register(observer3)

    fmt.Println("--- 第一次数据更新 ---")
    subject.SetData("Hello World!")
    time.Sleep(100 * time.Millisecond) // 等待goroutine完成

    // 注销一个观察者
    subject.Deregister(observer2)

    fmt.Println("--- 第二次数据更新 ---")
    subject.SetData(12345)
    time.Sleep(100 * time.Millisecond) // 等待goroutine完成

    // 再次注册一个观察者
    subject.Register(&ConcreteObserver{ID: "ObserverD"})

    fmt.Println("--- 第三次数据更新 ---")
    subject.SetData(map[string]string{"key": "value", "status": "active"})
    time.Sleep(100 * time.Millisecond) // 等待goroutine完成
}

DataSubject

sync.RWMutex

observers

data

Notify

Update

observer.Update(data)

go

Notify

observer.Update

// 示例：使用通道作为通知机制
type ChannelObserver struct {
    ID    string
    Ch    chan interface{} // 每个观察者有自己的输入通道
    Done  chan struct{}    // 用于停止观察者
}

func NewChannelObserver(id string) *ChannelObserver {
    o := &ChannelObserver{
        ID:   id,
        Ch:   make(chan interface{}, 10), // 缓冲通道
        Done: make(chan struct{}),
    }
    go o.Run() // 启动观察者处理循环
    return o
}

func (o *ChannelObserver) Run() {
    for {
        select {
        case data := <-o.Ch:
            fmt.Printf("通道观察者 %s 收到更新: %v\n", o.ID, data)
        case <-o.Done:
            fmt.Printf("通道观察者 %s 停止。\n", o.ID)
            return
        }
    }
}

func (o *ChannelObserver) Update(data interface{}) {
    // 将数据发送到自己的通道
    select {
    case o.Ch <- data:
        // 数据发送成功
    default:
        // 通道已满，可以记录日志或丢弃事件
        fmt.Printf("通道观察者 %s 的通道已满，丢弃事件: %v\n", o.ID, data)
    }
}

func (o *ChannelObserver) GetID() string {
    return o.ID
}

func (o *ChannelObserver) Stop() {
    close(o.Done)
}

// 在 DataSubject 的 Notify 方法中，可以这样调用：
// for _, observer := range s.observers {
//     observer.Update(data) // 如果observer是ChannelObserver，它会把数据发到自己的通道
// }

Notify

Notify

goroutine

Update

goroutine

Update