怎样用Golang实现发布订阅模式 基于channel构建事件驱动系统

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

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

// EventBus 事件总线
type EventBus struct {
    subscribers map[string][]chan EventData
    rm          sync.RWMutex
}

// NewEventBus 创建一个新的事件总线
func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]chan EventData),
    }
}

// Subscribe 订阅事件
func (bus *EventBus) Subscribe(eventType string, ch chan EventData) {
    bus.rm.Lock()
    defer bus.rm.Unlock()

    if _, found := bus.subscribers[eventType]; !found {
        bus.subscribers[eventType] = []chan EventData{}
    }
    bus.subscribers[eventType] = append(bus.subscribers[eventType], ch)
}

// Publish 发布事件
func (bus *EventBus) Publish(event EventData) {
    bus.rm.RLock()
    defer bus.rm.RUnlock()

    if chans, found := bus.subscribers[event.EventType]; found {
        for _, ch := range chans {
            // 使用go routine避免阻塞发布者
            go func(channel chan EventData, event EventData) {
                channel <- event
            }(ch, event)
        }
    }
}

// Unsubscribe 取消订阅
func (bus *EventBus) Unsubscribe(eventType string, ch chan EventData) {
    bus.rm.Lock()
    defer bus.rm.Unlock()

    if chans, found := bus.subscribers[eventType]; found {
        for i, channel := range chans {
            if channel == ch {
                bus.subscribers[eventType] = append(chans[:i], chans[i+1:]...)
                close(ch) // 关闭channel
                return
            }
        }
    }
}

func main() {
    rand.Seed(time.Now().UnixNano())

    bus := NewEventBus()

    // 创建订阅者 channels
    orderCreatedChan := make(chan EventData, 10)
    paymentProcessedChan := make(chan EventData, 10)
    shippingStartedChan := make(chan EventData, 10)

    // 订阅事件
    bus.Subscribe("order.created", orderCreatedChan)
    bus.Subscribe("payment.processed", paymentProcessedChan)
    bus.Subscribe("shipping.started", shippingStartedChan)

    // 启动消费者 goroutine
    go func() {
        for event := range orderCreatedChan {
            fmt.Printf("Order Created: %+v\n", event.Data)
        }
    }()

    go func() {
        for event := range paymentProcessedChan {
            fmt.Printf("Payment Processed: %+v\n", event.Data)
        }
    }()

    go func() {
        for event := range shippingStartedChan {
            fmt.Printf("Shipping Started: %+v\n", event.Data)
        }
    }()

    // 模拟发布事件
    for i := 0; i < 5; i++ {
        time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)

        switch rand.Intn(3) {
        case 0:
            bus.Publish(EventData{EventType: "order.created", Data: map[string]interface{}{"order_id": i + 1, "user_id": rand.Intn(100)}})
        case 1:
            bus.Publish(EventData{EventType: "payment.processed", Data: map[string]interface{}{"order_id": i + 1, "amount": float64(rand.Intn(1000)) / 100}})
        case 2:
            bus.Publish(EventData{EventType: "shipping.started", Data: map[string]interface{}{"order_id": i + 1, "tracking_number": "TRACK" + fmt.Sprintf("%d", rand.Intn(10000))}})
        }
    }

    // 等待一段时间，确保所有事件都被处理
    time.Sleep(2 * time.Second)

    // 取消订阅一个事件
    bus.Unsubscribe("payment.processed", paymentProcessedChan)
    fmt.Println("Unsubscribed from payment.processed events")

    // 发布更多的事件，观察是否还有payment.processed事件被处理
    for i := 5; i < 8; i++ {
        time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
        bus.Publish(EventData{EventType: "payment.processed", Data: map[string]interface{}{"order_id": i + 1, "amount": float64(rand.Intn(1000)) / 100}})
    }

    // 再次等待一段时间
    time.Sleep(1 * time.Second)

    fmt.Println("Done")
}

go func() {
    for event := range orderCreatedChan {
        orderData, ok := event.Data.(map[string]interface{})
        if !ok {
            fmt.Println("Invalid order data")
            continue
        }
        if orderData["user_id"].(int) > 50 { // 过滤user_id大于50的订单
            fmt.Printf("Order Created (Filtered): %+v\n", event.Data)
        }
    }
}()

Publish