NSQ学习:流控的实现

imxyb · · 1013 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

消息中间件的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()
}

参考链接


有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:imxyb

查看原文:NSQ学习:流控的实现

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

1013 次点击  
加入收藏 微博
上一篇:Golang-defer
下一篇:Go 字符串处理
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传