来自Go语言高级编程1.6章节
生产者消费者模型、发布订阅模型。
生产者消费者模型代码如下：
|
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 |
package main import ( "fmt" "os" "os/signal" "syscall" "time" ) 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) go Producer(5, ch) go Consumer(ch) time.Sleep(5 * time.Second) // Ctrl+C 退出程序 sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) fmt.Printf("quit (%v)\n", <-sig) } |
发布订阅模型需要先构建一个pubsub的发布订阅模型支持包：
|
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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
//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 //发布超时时间 subscriber map[subscriber]topicFunc //订阅者信息 } //创建一个发布者对象, 可以设置发布超时时间和缓存队列的长度 func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher { return &Publisher{ buffer: buffer, timeout: publishTimeout, subscriber: 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.subscriber[ch] = topic p.m.Unlock() return ch } //退出订阅 func (p *Publisher) Evict(sub chan interface{}) { p.m.Lock() defer p.m.Unlock() delete(p.subscriber, 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.subscriber { 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.subscriber { delete(p.subscriber, 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): } } |
然后实现一个本地的发布订阅模型使用程序：
|
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 |
package main import ( "AdvancedGo/ch1/pubsub" "fmt" "strings" "time" ) 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) } |
大致就是这样了。
【Go】常见的并发模式