在之前的学习中，接触到了消息队列的一些知识，然后今天在看书的时候，发现了一个使用go channel完成了一个订阅者模式的并发demo。

订阅者模式

关于订阅者模式，在这里给大家讲解一下，订阅者模式其实就是将产生的消息，给多个订阅者，然后订阅者可以看到这里订阅的消息。
这里实现的时候呢，因为有并发的知识，所以我们用到了协程，下面我们来具体的实现一下。

具体的实现流程

创建订阅者结构

type (
	subscriber chan interface{}         //订阅者的通道
	topicFunc  func(v interface{}) bool //主题过滤器
)

//发布者对象
type Publisher struct {
	m           sync.Mutex               //读写锁
	buffer      int                      //订阅队列缓存大小
	timeout     time.Duration            //发布者超时时间
	subscribers map[subscriber]topicFunc //订阅者消息
}

上面我们定义了基本的订阅者的结构，上面也有对订阅者字段的解释，这里着重理解的东西就是：

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) Publish(v interface{}) {
	p.m.Lock()
	defer p.m.Unlock()
	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) 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):
	}
}

这里我们使用到的并发式的添加消息，这里使用到了sync.WaitGroup，进行协程的同步。

关闭订阅者模式

//关闭发布对象，同时关闭
func (p *Publisher) Close() {
	p.m.Lock()
	defer p.m.Unlock()
	for sub := range p.subscribers {
		delete(p.subscribers, sub)
		close(sub)
	}
}

最后的测试

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,lzz")
	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)
}

最后的输出结果：