golang并发-发布订阅并发模型

￡神魔★判官ぃ 2022-09-17 03:03 161阅读 0赞

### 文章目录 ###

*  简介
 *  发布订阅模型

# 简介 #

发布订阅模型简写为 pub/sub 模型，消息生产者成为了发布者 publisher，消息消费者成为了订阅者 subscriber。传统生产者消费者模型是将消息发送到一个队列中，发布订阅模型是将消息发布给一个主题

与其把发布订阅者模式我更喜欢理解成发布订阅器模式，订阅者更像是一个订阅器，发布者中含有多个订阅器，每个订阅器有自己的一种订阅规则，同时订阅器中会存在多个不同的订阅信息，但是这些订阅信息都是满足该订阅器的订阅规则的

# 发布订阅模型 #

发布订阅支持包

package pubsub
    
    import (
        "sync"
        "time"
    )
    
    /* ============== type ============== */
    
    type (
    	// subscriber 是一个管道，实际是存储了一批订阅信息的信道，即「一批订阅信息的合集」
        subscriber chan interface{
        }       
      	// topicFunc 是一个函数，实际就是一个「订阅规则」
        topicFunc  func(v interface{
        }) bool
    )
    
    /* ============== struct ============== */
    
    // 发布者对象
    type Publisher struct {
        
      	// 读写锁
        m           sync.RWMutex
    	// 订阅器的订阅信息信道满载后的超时时延
        timeout     time.Duration
      
      	// 订阅器中可以容纳多少条订阅信息
        buffer      int                     
      	// 订阅器是一个键值对，key 是订阅信息信道，value 是订阅规则函数。发布者中含有无数个订阅器
        subscribers map[subscriber]topicFunc
    }
    
    /* ------------ 发布者新增一个订阅器 ------------ */
    
    // 发布者新增一个新的订阅器，入参是新的订阅过滤器的过滤规则
    func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{
        } {
        
        ch := make(chan interface{
        }, p.buffer)
    	// 加读锁
        p.m.Lock()
      	// 这样一个订阅器就创建了。订阅器是一个键值对，key 为订阅信息信道，value 为订阅器规则
        p.subscribers[ch] = topic
        p.m.Unlock()
        return ch
    }
    
    // 发布者添加一个新的订阅器，新的订阅没有订阅规则
    func (p *Publisher) Subscribe() chan interface{
        } {
        
    	// nil 即没有订阅规则
        return p.SubscribeTopic(nil)
    }
    
    /* ------------ 发布者删除一个订阅器 ------------ */
    
    // 删除一个订阅器
    func (p *Publisher) Evict(sub chan interface{
        }) {
        
      	// 因为涉及修改发布者 p 的内容，所以加写锁
        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()
    
      	// sync.WaitGroup 保证当发布者内每一个订阅器都和订阅信息 v 做完匹配之后 Publish 发布函数才能结束，否则卡在 Wait 函数处
        var wg sync.WaitGroup
        for sub, topic := range p.subscribers {
        
            wg.Add(1)
          	// 每将一个订阅器和订阅信息 v 做匹配就开启一个新的协程
            go p.sendTopic(sub, topic, v, &wg)
        }
        wg.Wait()
    }
    
    // 订阅器匹配订阅信息的核心内容，订阅器的过滤规则筛掉了发布过来的订阅信息就直接 return，如果没有筛掉就会考虑给订阅器添加这个订阅信息，如果在添加时发现该订阅器的订阅信息信道被存满了，就会进行超时等待
    func (p *Publisher) sendTopic(
        sub subscriber, topic topicFunc, v interface{
        }, wg *sync.WaitGroup,
    ) {
        
        defer wg.Done()
    	// 如果订阅信息不满足订阅器的过滤器规则，即被过滤器筛掉了，就直接 return
        if topic != nil && !topic(v) {
        
            return
        }
    
      	// 如果订阅信息没有被过滤器筛掉
        select {
        
        // 将订阅信息 v 存入订阅器的订阅信息信道内
        case sub <- v:
        // 如果该订阅器的订阅信息信道已满，则使用超时等待
        case <-time.After(p.timeout):
        }
    }
    
    /* ------------ 发布者清空其中的每个订阅器 ------------ */
    // 完全清空发布者中的每一个订阅器
    func (p *Publisher) Close() {
        
        // 因为是修改(删除)订阅器，所以需要加写锁
        p.m.Lock()
        defer p.m.Unlock()
    
        for sub := range p.subscribers {
        
            delete(p.subscribers, sub)
            close(sub)
        }
    }
    
    /* ============== create ============== */
    // 创建一个发布者，里头主要包含了发布者的读写锁以及无数个订阅器
    func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {
        
        return &Publisher{
        
    		// 订阅器中可以容纳多少条订阅信息
            buffer: buffer,
         	// 订阅器的订阅信息信道满载后的超时时延
            timeout: publishTimeout,
          	// 订阅器是一个键值对，key 是订阅信息信道，value 是订阅规则函数。发布者中含有无数个订阅器
            subscribers: make(map[subscriber]topicFunc),
        }
    }

下面是发布订阅 main 主流程

import "pubsub"
    
    func main() {
        
      	/* ============== 创建发布者 ============== */
      
      	// 创建发布者，其中规定了超时时延以及其中每一个订阅器的订阅信息满载数量
    	p := pubsub.NewPublisher(100 * time.Millisecond, 10) 
      	// main 最后清空发布者 p 中所有的订阅器
    	defer p.Close()
      
       	/* ============== 发布者新增订阅器 ============== */
      
    	// 发布者 p 新增一个订阅器，并且该订阅器中没有任何过滤规则，返回的是该订阅器的订阅信息信道
      	allMsg := p.Subscribe()
      	// 发布者 p 新增一个订阅器，并且该订阅器中有一个需要匹配字符串“golang”的订阅规则，返回的是该订阅器的订阅信息信道
        golang := p.SubscribeTopic(func(v interface{
        }) bool {
        
            if s, ok := v.(string); ok {
        
    			return strings.Contains(s, "golang")
        	}
        	return false
        })
    	// todo: 后面还可以添加多个订阅器
      
       	/* ============== 发布者发布信息 ============== */
      
      	// 开启一个协程发布，发布者发布一个订阅信息，这个订阅信息会和发布者内的所有订阅器去做匹配
      	go p.Publish("hello,  world!")
    	// 开启一个协程发布，发布者发布一个订阅信息，这个订阅信息会和发布者内的所有订阅器去做匹配
      	go p.Publish("hello, golang!")
    	// todo: 后面还可以发布多个订阅信息
      
      	/* ============== 打印发布者中订阅器的所有订阅信息 ============== */
      	
      	// 从信道中取数据，避免信道满载，打印无过滤规则的订阅器中的所有订阅信息，
      	go func() {
        
            for  msg := range allMsg {
        
                fmt.Println("allMsg:", msg)
            }
        }()
      	// 从信道中取数据，避免信道满载，打印有过滤规则("golang")的订阅器中的所有订阅信息
      	go func() {
        
            for  msg := range golangMsg {
        
                fmt.Println("golangMsg:", msg)
            }
        }()
      
       	/* ============== 等待时常 ============== */
      
      	// 2s 延时
    	time.Sleep(2 * time.Second)  
    }