如何用Golang实现发布订阅模式 使用channel构建事件驱动架构

package main

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

// EventData 事件数据结构
type EventData struct {
    Topic string
    Data  interface{}
}

// Broker 发布者/代理
type Broker struct {
    mu          sync.RWMutex
    subscribers map[string][]chan EventData
    quit        chan bool
}

// NewBroker 创建一个新的 Broker 实例
func NewBroker() *Broker {
    return &Broker{
        subscribers: make(map[string][]chan EventData),
        quit:        make(chan bool),
    }
}

// Subscribe 订阅特定主题
func (b *Broker) Subscribe(topic string, ch chan EventData) {
    b.mu.Lock()
    defer b.mu.Unlock()
    b.subscribers[topic] = append(b.subscribers[topic], ch)
}

// Unsubscribe 取消订阅
func (b *Broker) Unsubscribe(topic string, ch chan EventData) {
    b.mu.Lock()
    defer b.mu.Unlock()

    if _, ok := b.subscribers[topic]; !ok {
        return
    }

    var newSubs []chan EventData
    for _, sub := range b.subscribers[topic] {
        if sub != ch {
            newSubs = append(newSubs, sub)
        }
    }
    b.subscribers[topic] = newSubs

    // 如果该主题没有订阅者，则删除该主题
    if len(b.subscribers[topic]) == 0 {
        delete(b.subscribers, topic)
    }
}

// Publish 发布事件
func (b *Broker) Publish(event EventData) {
    b.mu.RLock()
    defer b.mu.RUnlock()

    if subs, ok := b.subscribers[event.Topic]; ok {
        for _, ch := range subs {
            // 使用 goroutine 防止阻塞发布者
            go func(channel chan EventData) {
                select {
                case channel <- event: // 发送事件到channel
                default:
                    fmt.Println("Channel full, dropping message") // 处理channel阻塞的情况
                }
            }(ch)
        }
    }
}

// Start 启动 Broker
func (b *Broker) Start() {
    <-b.quit // 等待退出信号
}

// Stop 停止 Broker
func (b *Broker) Stop() {
    close(b.quit)
}

func main() {
    broker := NewBroker()
    go broker.Start()
    defer broker.Stop()

    // 创建订阅者
    subscriber1 := make(chan EventData, 10) // 带缓冲的channel
    subscriber2 := make(chan EventData, 10)

    // 订阅主题
    broker.Subscribe("topicA", subscriber1)
    broker.Subscribe("topicB", subscriber2)

    // 发布事件
    broker.Publish(EventData{Topic: "topicA", Data: "Message for topicA"})
    broker.Publish(EventData{Topic: "topicB", Data: "Message for topicB"})
    broker.Publish(EventData{Topic: "topicA", Data: "Another message for topicA"})

    time.Sleep(time.Second) // 等待消息处理

    // 从订阅者接收消息
    select {
    case msg := <-subscriber1:
        fmt.Printf("Subscriber 1 received: %+v\n", msg)
    default:
        fmt.Println("Subscriber 1 received nothing")
    }

    select {
    case msg := <-subscriber1:
        fmt.Printf("Subscriber 1 received: %+v\n", msg)
    default:
        fmt.Println("Subscriber 1 received nothing")
    }

    select {
    case msg := <-subscriber2:
        fmt.Printf("Subscriber 2 received: %+v\n", msg)
    default:
        fmt.Println("Subscriber 2 received nothing")
    }


    // 取消订阅
    broker.Unsubscribe("topicA", subscriber1)

    // 再次发布事件，subscriber1 不应该再收到
    broker.Publish(EventData{Topic: "topicA", Data: "This should not be received by subscriber1"})
    time.Sleep(time.Second)

    select {
    case msg := <-subscriber1:
        fmt.Printf("Subscriber 1 received: %+v\n", msg)
    default:
        fmt.Println("Subscriber 1 received nothing") // 预期输出
    }
}

select

Publish