标签:print 发送 type 关系 pac lse 生活 关闭 生产者
作者:Gundy_
链接:https://www.jianshu.com/p/dc94f2099277
// 生产者: 生成 factor 整数倍的序列 func Producer(factor int, out chan<- int) { for i := 0; ; i++ { out <- i*factor } } // 消费者 func Consumer(in <-chan int) { for v := range in { fmt.Println(v) } } func main() { ch := make(chan int, 64) // 成果队列 go Producer(3, ch) // 生成 3 的倍数的序列 go Producer(5, ch) // 生成 5 的倍数的序列 go Consumer(ch) // 消费 生成的队列 // 运行一定时间后退出 time.Sleep(5 * time.Second) }
我们开启了2个Producer
生产流水线,分别用于生成3和5的倍数的序列。然后开启1个Consumer
消费者线程,打印获取的结果。我们通过在main
函数休眠一定的时间来让生产者和消费者工作一定时间。正如前面一节说的,这种靠休眠方式是无法保证稳定的输出结果的。
我们可以让main
函数保存阻塞状态不退出,只有当用户输入Ctrl-C
时才真正退出程序:
func main() { ch := make(chan int, 64) // 成果队列 go Producer(3, ch) // 生成 3 的倍数的序列 go Producer(5, ch) // 生成 5 的倍数的序列 go Consumer(ch) // 消费 生成的队列 // Ctrl+C 退出 sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) fmt.Printf("quit (%v)\n", <-sig) }
我们这个例子中有2个生产者,并且2个生产者之间并无同步事件可参考,它们是并发的。因此,消费者输出的结果序列的顺序是不确定的,这并没有问题,生产者和消费者依然可以相互配合工作。
发布订阅(publish-and-subscribe)模型通常被简写为pub/sub模型。在这个模型中,消息生产者成为发布者(publisher),而消息消费者则成为订阅者(subscriber),生产者和消费者是M:N的关系。在传统生产者和消费者模型中,是将消息发送到一个队列中,而发布订阅模型则是将消息发布给一个主题。
为此,我们构建了一个名为pubsub
的发布订阅模型支持包:
// Package pubsub implements a simple multi-topic pub-sub library. package pubsub import ( "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) Subscribe() chan interface{} { return p.SubscribeTopic(nil) } // 添加一个新的订阅者,订阅过滤器筛选后的主题 func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{} { ch := make(chan interface{}, p.buffer) p.m.Lock() p.subscribers[ch] = topic p.m.Unlock() return ch } // 退出订阅 func (p *Publisher) Evict(sub chan interface{}) { p.m.Lock() defer p.m.Unlock() delete(p.subscribers, sub) close(sub) } // 发布一个主题 func (p *Publisher) Publish(v interface{}) { p.m.RLock() defer p.m.RUnlock() var wg sync.WaitGroup for sub, topic := range p.subscribers { wg.Add(1) go p.sendTopic(sub, topic, v, &wg) } wg.Wait() } // 关闭发布者对象,同时关闭所有的订阅者管道。 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) 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): } }
下面的例子中,有两个订阅者分别订阅了全部主题和含有"golang"的主题:
import "path/to/pubsub" func main() { p := pubsub.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) }
标签:print 发送 type 关系 pac lse 生活 关闭 生产者
原文地址:https://www.cnblogs.com/-wenli/p/12336965.html