2022-07-10 分类：GO 作者：admin 阅读（25）
1.事件总线
事件总线是发布/订阅模式的实现，其中发布者发布数据，并且订阅者可以监听这些数据并基于这些数据作出处理。这使发布者与订阅者松耦合。发布者将数据事件发布到事件总线，总线负责将它们发送给订阅者。
传统的实现事件总线的方法会涉及到使用回调。订阅者通常实现接口，然后事件总线通过接口传播数据。
使用 Go 的并发模型，大多数地方可以使用 channel 来替代回调。
2.事件总线实例
EvevtBus.go
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
package EventBus import "sync" // 定义数据结构 type DataEvent struct { Data interface{} Topic string } // DataChannel是一个能接收 DataEvent 的 channel type DataChannel chan DataEvent // DataChannelSlice 是一个包含 DataChannels 数据的切片 type DataChannelSlice []DataChannel // 定义事件总线 EventBus 存储有关订阅者感兴趣的特定主题的信息 type EventBus struct { Subscribers map[string]DataChannelSlice rm sync.RWMutex } // 发布主题 发布者需要提供广播给订阅者所需要的主题和数据 func (eb *EventBus) Publish(topic string, data interface{}) { eb.rm.RLock() if chans, found := eb.Subscribers[topic]; found { // 这样做是因为切片引用相同的数组，即使它们是按值传递的 // 因此我们正在使用我们的元素创建一个新切片，从而正确地保持锁定 channels := append(DataChannelSlice{}, chans...) //切片赋值 //使用Goroutine 来避免阻塞发布者 go func(data DataEvent, dataChannelSlices DataChannelSlice) { for _, ch := range dataChannelSlices { ch <- data } }(DataEvent{Data: data, Topic: topic}, channels) } eb.rm.RUnlock() } // 订阅主题 如传统方法回调一样。当发布者向主题发布数据时，channel将接收数据。 func (eb *EventBus) Subscribe(topic string, ch DataChannel) { eb.rm.Lock() if prev, found := eb.Subscribers[topic]; found { eb.Subscribers[topic] = append(prev, ch) } else { eb.Subscribers[topic] = append([]DataChannel{}, ch) } eb.rm.Unlock() } |
main.go调用
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
package main import ( "Alang/EventBus/EventBus" "fmt" "math/rand" "time" ) // 声明事件总线对象 var eb = &EventBus.EventBus{ Subscribers: map[string]EventBus.DataChannelSlice{}, } //打印订阅消息 func printDataEvent(ch string, data EventBus.DataEvent) { fmt.Printf("Channel: %s; Topic: %s; DataEvent: %v\n", ch, data.Topic, data.Data) } //发布消息 func publishTo(topic string, data string) { for { eb.Publish(topic, data) time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) } } func main() { ch1 := make(chan EventBus.DataEvent) ch2 := make(chan EventBus.DataEvent) ch3 := make(chan EventBus.DataEvent) eb.Subscribe("topic1", ch1) eb.Subscribe("topic2", ch2) eb.Subscribe("topic3", ch3) go publishTo("topic1", "Welcome to topic-1") go publishTo("topic2", "Welcome to topic-2") for { select { case d := <-ch1: go printDataEvent("ch1", d) case d := <-ch2: go printDataEvent("ch2", d) case d := <-ch3: go printDataEvent("ch3", d) } } } |
运行结果
局限性：channel 如果没有订阅者消费，会阻塞原文链接：https://blog.csdn.net/finghting321/article/details/103394274
推荐阅读：http://t.zoukankan.com/Jun10ng-p-13173368.html
「三年博客，如果觉得我的文章对您有用，请帮助本站成长」
上一篇：golang中使用GOPATH模式和GoModule（gomod）模式的区别
分别用不同厚度的筏板定义，画图后这设置筏板变截面处理。 http://f.fwxgx.co...
评：新文章！分别用不同厚度的筏板定义，画图后这设置筏板变截面处理。 http://f.fwxgx.co...
评：新文章！新增一个框架图！ http://biji.jinli.vip/wp-content/upl...
评：新文章！嗨，这是一条评论。 要开始审核、编辑及删除评论，请访问仪表盘的“评论”页面。 评论者头像来自...
评：世界，您好！
共有 0 - Golang—实现简单的事件总线(发布订阅模式)