Golang发布订阅模式 channel实现方案-Golang-PHP中文网

Golang发布订阅模式 channel实现方案

P粉602998670

发布： 2025-08-23 11:14:01

原创

896人浏览过

Go语言中通过channel和goroutine实现发布订阅模式，核心角色为发布者、订阅者和消息中心。使用chan传递消息，消息中心管理订阅关系并广播消息，支持多主题、动态增删订阅者，利用缓冲channel避免阻塞，确保高效并发。示例中定义Topic、Event、Subscriber及PubSub结构，通过Start启动事件循环，Subscribe注册订阅，Publish发送消息，Unsubscribe移除订阅者，Stop停止服务。每个订阅者独立接收消息，发布者将事件推送到对应主题的channel，所有订阅该主题的订阅者均可收到副本。通过select非阻塞发送，防止慢订阅者影响整体性能，适用于事件通知、消息广播等场景。实际应用可扩展优先级、超时、持久化机制，需注意channel缓冲大小与资源释放，避免内存泄漏。整体结构清晰，并发安全，适合构建轻量级事件系统。

golang发布订阅模式 channel实现方案

在Go语言中，发布订阅模式可以通过

channel

登录后复制

和

goroutine

登录后复制

高效实现。这种模式适用于消息广播、事件通知等场景，核心是将消息从一个发布者分发给多个订阅者。

基本设计思路

发布订阅系统包含三个主要角色：发布者（Publisher）、订阅者（Subscriber）和消息中心（Broker）。使用

channel

登录后复制

作为消息传递的载体，通过一个中心化的结构管理订阅关系，并将每条消息广播给所有活跃的订阅者。

关键点：

使用
```
chan interface{}
```
登录后复制
或具体类型的
```
channel
```
登录后复制
传递消息
每个订阅者拥有独立的接收
```
channel
```
登录后复制
消息中心负责复制消息到每个订阅者的
```
channel
```
登录后复制
支持动态添加和移除订阅者

简单实现示例

// Topic 表示订阅主题 type Topic string

// Event 消息结构 type Event struct { Topic Topic Data interface{} }

// Subscriber 订阅者定义 type Subscriber chan Event

// NewSubscriber 创建新订阅者 func NewSubscriber() Subscriber { return make(Subscriber, 10) // 缓冲channel避免阻塞 }

// 关闭订阅者 func (s Subscriber) Close() { close(s) }

// PubSub 消息中心 type PubSub struct { subscribers map[Topic][]Subscriber register chan struct{} unregister chan Subscriber publish chan Event quit chan struct{} }

// NewPubSub 创建发布订阅系统 func NewPubSub() *PubSub { return &PubSub{ subscribers: make(map[Topic][]Subscriber), register: make(chan struct{}), unregister: make(chan Subscriber), publish: make(chan Event), quit: make(chan struct{}), } }

// Start 启动消息分发循环 func (ps *PubSub) Start() { for { select { case event := <-ps.publish: if subs, ok := ps.subscribers[event.Topic]; ok { for _, sub := range subs { select { case sub <- event: default: // 避免因订阅者阻塞影响其他分发 } } } case sub := <-ps.unregister: for topic, subscribers := range ps.subscribers { for i, s := range subscribers { if s == sub { ps.subscribers[topic] = append(subscribers[:i], subscribers[i+1:]...) break } } } sub.Close() case <-ps.quit: return } } }

// Subscribe 订阅主题 func (ps *PubSub) Subscribe(topic Topic) Subscriber { sub := NewSubscriber() ps.register <- struct{}{} // 占位（可扩展注册逻辑） if _, ok := ps.subscribers[topic]; !ok { ps.subscribers[topic] = []Subscriber{} } ps.subscribers[topic] = append(ps.subscribers[topic], sub) return sub }

// Publish 发布消息 func (ps *PubSub) Publish(topic Topic, data interface{}) { ps.publish <- Event{Topic: topic, Data: data} }

// Unsubscribe 取消订阅 func (ps *PubSub) Unsubscribe(sub Subscriber) { ps.unregister <- sub }

// Stop 停止服务 func (ps *PubSub) Stop() { close(ps.quit) }

使用示例

func main() { pubsub := NewPubSub() go pubsub.Start()

// 订阅新闻主题
sub1 := pubsub.Subscribe("news")
sub2 := pubsub.Subscribe("news")

// 订阅天气主题
sub3 := pubsub.Subscribe("weather")

// 模拟接收消息
go func() {
    for event := range sub1 {
        fmt.Printf("用户1收到: %v\n", event.Data)
    }
}()

go func() {
    for event := range sub2 {
        fmt.Printf("用户2收到: %v\n", event.Data)
    }
}()

go func() {
    for event := range sub3 {
        fmt.Printf("用户3收到天气: %v\n", event.Data)
    }
}()

// 发布消息
pubsub.Publish("news", "今日全球要闻")
pubsub.Publish("weather", "北京晴，25℃")

time.Sleep(time.Second)
pubsub.Unsubscribe(sub1) // 取消订阅

pubsub.Publish("news", "突发新闻更新") // sub2 仍能收到

time.Sleep(time.Second)
pubsub.Stop()

登录后复制

}