Golangchannel实现发布订阅系统

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>

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