Golang 消息/事件发布订阅模式实现 - Hub 库使用指南

消息/事件发布订阅(Pub/Sub)模式是一种常见的解耦设计模式，在Go中可以通过多种方式实现。下面我将介绍一个简单高效的Hub库实现及使用方法。

基本概念

发布订阅模式包含三个核心组件：

1. 发布者(Publisher)：负责发布消息/事件
2. 订阅者(Subscriber)：注册对特定消息/事件的兴趣
3. 消息中心(Hub)：负责接收消息并分发给订阅者

简单Hub实现

以下是一个线程安全的简单Hub实现：

package hub

import (
	"sync"
)

type Message interface{}

type Hub struct {
	subscribers map[string][]chan Message
	mu          sync.RWMutex
}

func NewHub() *Hub {
	return &Hub{
		subscribers: make(map[string][]chan Message),
	}
}

// 订阅主题
func (h *Hub) Subscribe(topic string) chan Message {
	h.mu.Lock()
	defer h.mu.Unlock()

	ch := make(chan Message, 100) // 缓冲通道防止阻塞
	h.subscribers[topic] = append(h.subscribers[topic], ch)
	return ch
}

// 发布消息
func (h *Hub) Publish(topic string, msg Message) {
	h.mu.RLock()
	defer h.mu.RUnlock()

	if subs, ok := h.subscribers[topic]; ok {
		for _, ch := range subs {
			ch <- msg
		}
	}
}

// 取消订阅
func (h *Hub) Unsubscribe(topic string, ch chan Message) {
	h.mu.Lock()
	defer h.mu.Unlock()

	if subs, ok := h.subscribers[topic]; ok {
		for i, sub := range subs {
			if sub == ch {
				h.subscribers[topic] = append(subs[:i], subs[i+1:]...)
				close(ch)
				break
			}
		}
	}
}

使用示例

package main

import (
	"fmt"
	"time"
)

func main() {
	hub := NewHub()

	// 订阅者1订阅"news"主题
	sub1 := hub.Subscribe("news")
	go func() {
		for msg := range sub1 {
			fmt.Printf("Subscriber 1 received: %v\n", msg)
		}
		fmt.Println("Subscriber 1 closed")
	}()

	// 订阅者2订阅"news"主题
	sub2 := hub.Subscribe("news")
	go func() {
		for msg := range sub2 {
			fmt.Printf("Subscriber 2 received: %v\n", msg)
		}
		fmt.Println("Subscriber 2 closed")
	}()

	// 订阅者3订阅"weather"主题
	sub3 := hub.Subscribe("weather")
	go func() {
		for msg := range sub3 {
			fmt.Printf("Subscriber 3 received: %v\n", msg)
		}
		fmt.Println("Subscriber 3 closed")
	}()

	// 发布消息
	hub.Publish("news", "Breaking news: Go 1.20 released!")
	hub.Publish("weather", "Today's weather: Sunny, 25°C")
	
	time.Sleep(time.Second) // 等待消息处理
	
	// 取消订阅
	hub.Unsubscribe("news", sub1)
	
	hub.Publish("news", "Another news: AI makes progress")
	
	time.Sleep(time.Second) // 等待消息处理
}

高级特性扩展

如果需要更强大的功能，可以考虑以下扩展：

1. 通配符订阅：支持类似"news.*"的通配符模式
2. 持久化：将消息持久化到数据库
3. 消息过滤：订阅时增加过滤条件
4. QoS保证：实现不同级别的消息传递保证

现有成熟库推荐

如果不想自己实现，可以考虑以下成熟的开源库：

1. go-eventbus：轻量级事件总线

   import "github.com/asaskevich/EventBus"
   
   bus := EventBus.New()
   bus.Subscribe("topic", func(msg string) { /* handler */ })
   bus.Publish("topic", "message")
   

2. emitter：高性能事件发射器

   import "github.com/olebedev/emitter"
   
   e := &emitter.Emitter{}
   e.On("event", func(event *emitter.Event) { /* handler */ })
   e.Emit("event", "data")
   

3. watermill：用于构建事件驱动应用的库

   import "github.com/ThreeDotsLabs/watermill"
   
   pubSub := watermill.NewStdPubSub(false, watermill.NopLogger{})
   messages, _ := pubSub.Subscribe(context.Background(), "topic")
   pubSub.Publish("topic", watermill.NewMessage("1", []byte("payload")))
   

最佳实践

1. 合理设置通道缓冲区大小
2. 确保订阅者及时处理消息，避免阻塞
3. 使用context实现优雅的取消订阅
4. 考虑消息序列化格式(JSON, Protobuf等)
5. 在高并发场景下注意性能优化

这个简单的Hub实现足以满足大多数基本发布订阅需求，对于更复杂的场景可以考虑使用上述成熟的第三方库。