go - 发布订阅模型

// 3.5 发布订阅模型
package mainimport ("fmt""strings""sync""time"
)type (subscriber chan interface{}         // 订阅者为一个通道topicFunc  func(v interface{}) bool // 主题为一个过滤器
)// 发布者对象
type Publisher struct {m           sync.RWMutex             //读写锁buffer      int                      // 订阅队列的缓存大小timeout     time.Duration            // 发布超时时间subscribers map[subscriber]topicFunc // 订阅者信息
}// 构建一个发布者对象，可以设置发布超时时间和缓存队列的长度
func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {return &Publisher{buffer:      buffer,timeout:     publishTimeout,subscribers: make(map[subscriber]topicFunc),}}// 关闭发布者对象，同时关闭所有的订阅通道
func (p *Publisher) Close() {p.m.Lock()defer p.m.Unlock()for sub := range p.subscribers {delete(p.subscribers, sub)close(sub)}
}// 添加一个新的订阅者，订阅过滤器筛选后的主题
func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{} {ch := make(chan interface{}, p.buffer)p.m.Lock()p.subscribers[ch] = topicp.m.Unlock()return ch
}// 添加一个新的订阅者，订阅全部主题
func (p *Publisher) Subscribe() chan interface{} {return p.SubscribeTopic(nil)
}// 退出订阅
func (p *Publisher) Evict(sub chan interface{}) {p.m.Lock()defer p.m.Unlock()delete(p.subscribers, sub)close(sub)
}// 发送主题,可以容忍一定的超时
func (p *Publisher) sendTopic(sub subscriber, topic topicFunc, v interface{}, wg *sync.WaitGroup,
) {defer wg.Done()if topic != nil && !topic(v) { // 过滤信息return}select {case sub <- v:case <-time.After(p.timeout): // 超时}
}// 发布一个主题
func (p *Publisher) Publish(v interface{}) {p.m.Lock()defer p.m.Unlock()var wg sync.WaitGroupfor sub, topic := range p.subscribers {wg.Add(1)go p.sendTopic(sub, topic, v, &wg)}wg.Wait()
}func main() {p := NewPublisher(100*time.Millisecond, 10)defer p.Close()all := p.Subscribe()golang := p.SubscribeTopic(func(v interface{}) bool {if s, ok := v.(string); ok {return strings.Contains(s, "golang")}return false})p.Publish("hello world")p.Publish("hello, golang")go func() {for msg := range all {fmt.Println("all:", msg)}}()go func() {for msg := range golang {fmt.Println("golang:", msg)}}()// 运行一段时间后退出time.Sleep(3 * time.Second)
}

go - 发布订阅模型相关推荐

kafka 发布订阅_在Kafka中发布订阅模型
kafka 发布订阅这是第四个柱中的一系列关于同步客户端集成与异步系统( 1, 2, 3 ). 在这里,我们将尝试了解Kafka的工作方式,以便正确利用其发布-订阅实现. 卡夫卡概念根据官方文件 ...
在Kafka中发布订阅模型
这是第四个柱中的一系列关于同步客户端集成与异步系统( 1, 2, 3 ). 在这里,我们将尝试了解Kafka的工作方式,以便正确利用其发布-订阅实现. 卡夫卡概念根据官方文件 : Kafka是一种分 ...
java实现rabbitmq发布/订阅模型（Publish/Subscribe queues），生产者消费者交换机消息队列
发布/订阅模型又称扇出模型,或者是广播模型,可以有多个消费者,每个消费者有自己的队列,每个队列都要绑定到交换机,生产者发送的消息只需要发送到交换机,再由交换机决定要发送到哪些队列,生产者无法自行决定. ...
分布式发布订阅模型网络的实现有哪些
大数据中常用的分布式发布订阅系统: 参考资料: WCF百科介绍分布式"消息发布者-订阅者"模型的实现--WCF双工通讯特性的应用 [设计模式] 观察者模式(发布-订阅/Publi ...
3，ActiveMQ-入门（基于JMS发布订阅模型）
一.Pub/Sub-发布/订阅消息传递模型在发布/订阅消息模型中,发布者发布一个消息,该消息通过topic传递给所有的客户端.在这种模型中,发布者和订阅者彼此不知道对方,是匿名的且可以动态发布和订阅 ...
ros 单向通讯 talker，listener 发布订阅模型
原文链接: ros 单向通讯 talker,listener 发布订阅模型上一篇: VirtualBox 端口转发(端口映射) 主机和虚拟机相互访问下一篇: python 串口编程发布订阅模型 ...
vue发布订阅模式,发布订阅模型
1.什么是发布订阅模式 (又叫做观察者模式) 他定义了对象间的一种一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于他的对象都将得到通知. 2.发布订阅模型 1.发布者会给订阅者提供一个方法以便 ...
c#事件的发布-订阅模型_C# 委托和事件与观察者模式(发布-订阅模式)讲解 by天命...
使用面向对象的思想用c#控制台代码模拟猫抓老鼠我们先来分析一下猫抓老鼠的过程 1.猫叫了 2.所有老鼠听到叫声,知道是哪只猫来了 3.老鼠们逃跑,边逃边喊:"xx猫来了,快跑啊!我是老鼠 ...
c#事件的发布-订阅模型_NET Core 3 WPF MVVM框架 Prism系列之事件聚合器
本文将介绍如何在.NET Core3环境下使用MVVM框架Prism的使用事件聚合器实现模块间的通信一.事件聚合器# 在上一篇 .NET Core 3 WPF MVVM框架 Prism系列之模块化 ...
c#事件的发布-订阅模型_C#事件订阅发布实现原理详解
我们用一个简单的例子,来说明一下这种消息传递的机制. 有一家三口,妈妈负责做饭,爸爸和孩子负责吃...将这三个人,想象成三个类. 妈妈有一个方法,叫做"做饭".有一个事件,叫做&q ...

go - 发布订阅模型

go - 发布订阅模型相关推荐

最新文章

热门文章