消息中间件的pull与push
消息中间件的实现无非两种套路，一种让客户端pull，典型的比如kafka便是如此，而另一种则是push，也就是让客户端不需要做任何操作，只需要做好conn便可以源源不断收到服务端的推送，典型的代表就是我们今天介绍的nsq。
pull的优势在于客户端可以自己做流控，比如客户端想什么时候pull就什么时候pull，不会因为服务端的强迫而接受，但劣势也很明显，如果服务端的生产速度很慢，客户端需要不断的轮询会让cpu处于繁忙且无用的状态。
push的优势则在于能够不受限于客户端的速度，可以让服务端更快的、批量的把数据push给客户端，因此大部分push实现的消息中间件都是属于内存型，而nsq比较特殊，它实际上是内存+磁盘的一个消息中间件。
push流的nsq如何做流控
上面也说了，pull流的优势在于可以让客户端自由控制消息的速度，但是push流不一样，push流不管客户端是否多繁忙都会推送消息，如果没有一个流控机制，很容易让客户端最终因为消费速度跟不上导致产生各种性能问题。nsq其实也考虑到这一点，于是采用了一个RDY的状态字段来表示流控。简单来说，就是客户端连接上nsqd之后，会告诉nsqd它的可接受的消息数量是多少，每当nsqd给客户端推送一条消息这个RDY就会减一，而客户端消费完毕并且发送一个FIN之后，这个RDY又会加一（其实这个设计有点类似tcp中的用来控制流量的窗口机制）
go-nsq客户端
我们来参考一下golang官方实现的nsq客户端是如何控制这个rdy的。
首先编写一个客户端：
type Customter struct {}
func (c *Customter) HandleMessage(msg *nsq.Message) error {
fmt.Println("receive: ", string(msg.Body))
return nil
}
func main() {
cfg := nsq.NewConfig()
cfg.LookupdPollInterval = time.Second
customer, err := nsq.NewConsumer("test", "t1", cfg)
if err != nil {
log.Panic(err)
}
customer.AddHandler(&Customter{})
if err := customer.ConnectToNSQD("127.0.0.1:4161"); err != nil {
log.Panic(err)
}
select {}
}
跳进源码，可以看到go-nsq的Consumer结构体有一个字段connections：
// github.com/nsqio/go-nsq/customer.go
type Consumer struct {
.....
connections map[string]*Conn
....
}
=
当我们上面的demo调用ConnectToNSQD的时候，这个connections的map会写入对应的nsqd addr作为key，连接成功的Conn作为value:
r.connections[addr] = conn
for _, c := range r.conns() {
r.maybeUpdateRDY(c)
}
上面代码表示会遍历这个Customer的所有nsqd conn（customer可以同时连接多个nsqd），然后调用maybeUpdateRDY这个方法：
// 当剩余rdy的数量等于1，或者少于最近一次的rdycount的25%，就调整这个rdycount，这个rdycount就取用户设置的MaxInFlight
if remain <= 1 || remain < (lastRdyCount/4) || (count > 0 && count < remain) {
r.log(LogLevelDebug, "(%s) sending RDY %d (%d remain from last RDY %d)",
conn, count, remain, lastRdyCount)
r.updateRDY(conn, count)
} else {
由此我们可以得知，nsqd的客户端在连接nsqd的时候就会设置一个初始的rdycount。当然，在连接成功之后，也会有一个gorountine后台不断去调整这个rdycount：
func (r *Consumer) rdyLoop() {
redistributeTicker := time.NewTicker(r.config.RDYRedistributeInterval)
for {
select {
case <-redistributeTicker.C:
r.redistributeRDY()
case <-r.exitChan:
goto exit
}
}
exit:
redistributeTicker.Stop()
r.log(LogLevelInfo, "rdyLoop exiting")
r.wg.Done()
}
参考链接
有疑问加站长微信联系（非本文作者）
