Golang发布订阅模式实现 channel事件系统-小浪学习网

答案：go语言通过channel和goroutine实现发布订阅模式，核心为EventBus管理topic与订阅者映射，发布者发送事件至指定topic，所有订阅该topic的订阅者通过channel接收事件，实现解耦通信。

Golang发布订阅模式实现 channel事件系统

在Go语言中，发布订阅模式（Pub/Sub）是一种常见的解耦组件通信的方式。通过使用 channel 和 goroutine，我们可以轻松实现一个轻量级、高效的事件系统。这种系统适用于事件驱动架构、消息广播、状态变更通知等场景。

基本设计思路

发布订阅系统的核心是将消息的发送者（发布者）与接收者（订阅者）解耦。所有订阅者监听同一个主题（topic），当某个主题有消息发布时，所有订阅该主题的消费者都会收到通知。

我们可以通过以下结构实现：

Event：表示一个事件，包含主题（topic）和数据（data）
Publisher：负责向指定 topic 发送事件
Subscriber：注册到某个 topic，接收并处理事件
EventBus：核心调度器，管理 topic 与 channel 的映射关系

使用 channel 实现事件总线

利用 channel 作为消息传递的媒介，每个 topic 对应一个或多个 channel。订阅者通过 channel 接收消息，发布者将消息发送到对应 channel。

立即学习“go语言免费学习笔记（深入）”；

下面是一个简单但实用的实现：

 package main <p>import ( "fmt" "sync" )</p><p>// Event 事件结构 type Event struct { Topic string Data  interface{} }</p><p>// Subscriber 订阅者类型，用于接收事件 type Subscriber chan Event</p><p>// NewSubscriber 创建一个新的订阅者 func NewSubscriber() Subscriber { return make(Subscriber, 10) // 缓冲 channel，避免阻塞 }</p><p>// 关闭订阅者 func (s Subscriber) Close() { close(s) }</p><p>// EventBus 事件总线 type EventBus struct { subscribers map[string][]Subscriber mutex       sync.RWMutex }</p><p>// NewEventBus 创建事件总线 func NewEventBus() *EventBus { return &EventBus{ subscribers: make(map[string][]Subscriber), } }</p><p>// Subscribe 订阅某个 topic func (bus *EventBus) Subscribe(topic string) Subscriber { bus.mutex.Lock() defer bus.mutex.Unlock()</p><p>sub := NewSubscriber() bus.subscribers[topic] = append(bus.subscribers[topic], sub) return sub }</p><p>// Publish 发布事件到指定 topic func (bus *EventBus) Publish(topic string, data interface{}) { bus.mutex.RLock() defer bus.mutex.RUnlock()</p><p>if subs, ok := bus.subscribers[topic]; ok { event := Event{Topic: topic, Data: data} for _, sub := range subs { select { case sub <- event: default: // 非阻塞发送，如果 channel 满则跳过，防止发布者被阻塞 } } } }</p><p>// Unsubscribe 取消订阅 func (bus *EventBus) Unsubscribe(topic string, sub Subscriber) { bus.mutex.Lock() defer bus.mutex.Unlock()</p><p>if subs, ok := bus.subscribers[topic]; ok { for i, subscriber := range subs { if subscriber == sub { // 移除订阅者 bus.subscribers[topic] = append(subs[:i], subs[i+1:]...) sub.Close() break } } } }</p>

使用示例

下面是一个简单的使用场景：两个订阅者监听 “news” 主题，当有新消息发布时，它们都会收到通知。

 func main() {   bus := NewEventBus() <p>// 订阅者1 sub1 := bus.Subscribe("news") go func() { for event := range sub1 { fmt.Printf("订阅者1 收到消息: %s -> %vn", event.Topic, event.Data) } }()</p><p>// 订阅者2 sub2 := bus.Subscribe("news") go func() { for event := range sub2 { fmt.Printf("订阅者2 收到消息: %s -> %vn", event.Topic, event.Data) } }()</p><p>// 发布消息 bus.Publish("news", "今天天气不错") bus.Publish("news", "Go语言真好用")</p><p>// 等待接收 // 实际中可用 waitGroup 或 time.Sleep 等方式控制生命周期 // 这里简单暂停一下 // 在真实项目中应妥善管理生命周期 }</p>