消息中间件的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()
}
参考链接
有疑问加站长微信联系(非本文作者)