监听DB配置变更之go-broadcast简单实现

1. 前言

之前遇到一个需求，因为配置的查找是基于db的，而db的更改却无法实时通知到具体利用到这条数据的使用方，为了实现db数据变动时，能够尽快让使用方知道这条数据发生了变更，从而进行后续数据变更等相关逻辑的运行，就需要实现db数据变动时的通知。

在观察者模式中，因为观察者模式是一种一对多的关系模式，即多个观察者观察同一个主题对象，当主题对象发生变化时，会通知所有的观察者对象。

2. 分析

使用观察者模式来实现的话，则需要实现如下四个部分的结构：

1. 抽象主题
2. 具体主题
3. 抽象观察者
4. 具体观察者

举个例子，在我们日常使用微信公众号中，当你关注了一个公众号，这个公众号如果有更新的话，则会推送给每一个关注过这个公众号的用户。此时我们可以将具体的部分的接收映射到微信公众号中，即：

1. 抽象主题：公众号，具备订阅、取消订阅和发送消息的功能
2. 具体主题：具体某一个公众号
3. 抽象观察者：用户(泛指使用微信公众号的用户受众)
4. 具体观察者：某一个具体的用户

分析了以上四个结构之后，我们需要实现的功能部分就清楚了。即我们需要实现一个抽象主题，这个主题需要有提供注册、取消注册以及提交信息的能力，当提交信息到抽象主题的时候，抽象主题需要将这个消息通知到所有已经注册过的具体观察者。

3. 实现

在明确了需求之后， 就开始进行功能的实现，因为使用的是go语言，则第一时间肯定是希望通过chan这样的功能来实现，因为chan天生具备监听的能力，我们可以通过监听注册到抽象主题的chan，从而实现抽象主题消息的实时监听。

但秉持着“你需要的功能，基本都有人实现过”的方针，第一时间还是上到了github，看看是否有现成的开源方案，经过一番查找，还真发现了一个开源库可以使用，这个库的名称是go-broadcast。

下面就来说下broadcaster是如何实现上面的功能逻辑的，broadcaster这个库的代码很简单，主体实现逻辑只有110行代码左右，但符合我们的功能逻辑实现需要。

type broadcaster struct {
	input chan interface{}
	reg   chan chan<- interface{}
	unreg chan chan<- interface{}

	outputs map[chan<- interface{}]bool
}

// The Broadcaster interface describes the main entry points to
// broadcasters.
type Broadcaster interface {
	// Register a new channel to receive broadcasts
	Register(chan<- interface{})
	// Unregister a channel so that it no longer receives broadcasts.
	Unregister(chan<- interface{})
	// Shut this broadcaster down.
	Close() error
	// Submit a new object to all subscribers
	Submit(interface{})
	// Try Submit a new object to all subscribers return false if input chan is fill
	TrySubmit(interface{}) bool
}

首先定义了一个接口叫做Broadcaster，然后定义了一个broadcaster实现了Broadcaster的所有方法逻辑。

func (b *broadcaster) Register(newch chan<- interface{}) {
	b.reg <- newch
}

func (b *broadcaster) Unregister(newch chan<- interface{}) {
	b.unreg <- newch
}

func (b *broadcaster) Close() error {
	close(b.reg)
	close(b.unreg)
	return nil
}

// Submit an item to be broadcast to all listeners.
func (b *broadcaster) Submit(m interface{}) {
	if b != nil {
		b.input <- m
	}
}

• Register方法主要实现了将注册的chan直接放入到reg这个chan中，用于后续注册
• Register方法主要实现了将注册的chan直接让如到ureg这个chan中，用于后续注销
• Close方法主要是关闭reg和ureg两个chan
• Submit方法主要实现对抽象主题broadcaster发送消息，将消息放入input这个chan中

上面的方法都是基于chan作为通信的，而chan中有了数据，后续需要消费数据。

// NewBroadcaster creates a new broadcaster with the given input
// channel buffer length.
func NewBroadcaster(buflen int) Broadcaster {
	b := &broadcaster{
		input:   make(chan interface{}, buflen),
		reg:     make(chan chan<- interface{}),
		unreg:   make(chan chan<- interface{}),
		outputs: make(map[chan<- interface{}]bool),
	}

	go b.run()

	return b
}

这里的run()方法则是消费所有chan数据的地方。

func (b *broadcaster) broadcast(m interface{}) {
	for ch := range b.outputs { // 遍历所有注册的chan，将消息发送到注册的chan中
		ch <- m
	}
}

func (b *broadcaster) run() {
	for {
		select {
		case m := <-b.input: // 如果有消息输入，则广播出去
			b.broadcast(m)
		case ch, ok := <-b.reg: // 如果有新注册的，则进行output的添加
			if ok {
				b.outputs[ch] = true
			} else {
				return
			}
		case ch := <-b.unreg: // 如果有注销的，则进行output的删除
			delete(b.outputs, ch)
		}
	}
}

整体的运行图如下：

• 对应chan通过reg进行注册，注册后的chan记录在outputs中
• 对应chan通过ureg进行注销，注销后的chan从output中移除
• 对应的信息通过input输入，输入后的msg通过遍历outputs注册列表，从而通知到每一个注册者

4. 问题

在使用go-broadcast的过程中，看到之前有个pr加了一个TrySubmit的逻辑，这个逻辑主要是解决当input被装满了以后，broadcast会被阻塞，这个时候如果有新的消息进来，如何办呢？

// TrySubmit attempts to submit an item to be broadcast, returning
// true iff it the item was broadcast, else false.
func (b *broadcaster) TrySubmit(m interface{}) bool {
	if b == nil {
		return false
	}
	select {
	case b.input <- m:
		return true
	default:
		return false
	}
}

解决办法是采用select的方法尝试去塞入，塞入不成功则意味着消息提交失败，返回false，让使用者根据消息提交的结果进行后续的逻辑处理。

但这里还存在另外一个问题，库中给了一个样本case，这个样本case基于的条件都是消息传递给chan的时候没有阻塞。如下代码所示：

func (b *broadcaster) broadcast(m interface{}) {
	for ch := range b.outputs {
		ch <- m
	}
}

但一旦有注册的chan消费的时候阻塞了，这时候就会产生问题，会导致其它正常消费的chan因为一个异常chan而全部被阻塞住，导致其他chan都无法正常消费。

这个时候就会导致在input没有满的时候，即消息可以放入，但是消息无法被正常的消费，进而又反向导致input逐渐被塞满，最终导致input无法被塞入，消息也无法被发送到对应的chan中，导致run方法逻辑卡在broadcast中，导致整个运行出现问题。

解决办法：

func (b *broadcaster) broadcast(m interface{}) {
	for ch := range b.outputs {
		// if exist one output consume the chan message is too slow,
		// will block other output receive the msg.
		select {
		case ch <- m:
		default:
		}
	}
}

但这种虽然解决了一个chan满消费block其他chan的问题，随之也引入了丢消息的问题了，即有些消费慢的chan，由于chan消费慢导致无法接收新的消息，进而导致新消息丢失的问题。

5. 小结

因为需要实时监听db配置的变更，所以去探寻了一下方案，最终采用了go-broadcast的方案，但在使用go-broadcast的过程中，发现在broadcast消息的时候存在阻塞的行为，为了保证整个服务不被某个chan阻塞而停止运行，在broadcast消息的时候添加了select default条件来规避这个问题。

6. 参考

• go-broadcast