NSQ原始碼-Nsq客戶端
摘要：
看完lookupd和nsqd之後我們再來看下nsq client端的程式碼。 我是想把nsq系統完完整整的看一遍，從而對他形成一個更整體的
認識。對message queue來說他的client端就是生產者和消費者，生產者負責想nsq中投遞訊息，消費者負責從lookupd中獲取到...
看完lookupd和nsqd之後我們再來看下nsq client端的程式碼。 我是想把nsq系統完完整整的看一遍，從而對他形成一個更整體的
認識。對message queue來說他的client端就是生產者和消費者，生產者負責想nsq中投遞訊息，消費者負責從lookupd中獲取到
指定nsqd之後，從nsqd中獲取訊息。
生產者
我們以nsq/apps/to_nsq/to_nsq.go為例，客戶端這邊的程式碼邏輯就簡單很多，NewProducer例項化一個instance，publish訊息
到nsqd。
/// nsq/apps/to_nsq/to_nsq.go producer, err := nsq.NewProducer(addr, cfg) err := producer.Publish(*topic, line)
下面來看下Publish裡的具體邏輯。
// Publish synchronously publishes a message body to the specified topic, returning
// an error if publish failed
func (w *Producer) Publish(topic string, body []byte) error {
// 生成具體的cmd
return w.sendCommand(Publish(topic, body))
}
func (w *Producer) sendCommand(cmd *Command) error {
doneChan := make(chan *ProducerTransaction)
err := w.sendCommandAsync(cmd, doneChan, nil)
if err != nil {
close(doneChan)
return err
}
t := <-doneChan
return t.Error
}
func (w *Producer) sendCommandAsync(cmd *Command, doneChan chan *ProducerTransaction,
args []interface{}) error {
// keep track of how many outstanding producers we're dealing with
// in order to later ensure that we clean them all up...
atomic.AddInt32(&w.concurrentProducers, 1)
defer atomic.AddInt32(&w.concurrentProducers, -1)
if atomic.LoadInt32(&w.state) != StateConnected {
// 這裡是一個lazily connect
err := w.connect()
if err != nil {
return err
}
}
t := &ProducerTransaction{
cmd:cmd,
doneChan: doneChan,
Args:args,
}
select {
case w.transactionChan <- t:
case <-w.exitChan:
return ErrStopped
}
return nil
}
在connect函式裡啟動了一個go routine去處理transactionChan對應的東西
func (w *Producer) connect() error {
w.closeChan = make(chan int)
w.conn = NewConn(w.addr, &w.config, &producerConnDelegate{w})
w.conn.SetLogger(logger, logLvl, fmt.Sprintf("%3d (%%s)", w.id))
_, err := w.conn.Connect()
w.wg.Add(1)
go w.router()
這裡需要注意一下， go-nsq/conn.go是對底層連線的一個抽象，他是不關心你是生產者還是消費者，這裡使用到了
delegate 模式，conn.go收到訊息的處理放到了producerConnDelegate和consumerConnDelegate中，然後通知到具體的
消費者活著生產者。
消費者
回過頭我們再來看下消費者部分的程式碼，client端我們以nsq/apps/nsq_tail/nsq_tail.go為例，程式碼的基本邏輯如下：
// 1. new comsunmer instanace
consumer, err := nsq.NewConsumer(topics[i], *channel, cfg)
// 2. add handler
consumer.AddHandler(&TailHandler{topicName: topics[i], totalMessages: *totalMessages})
// 3. connect to nsqd
consumer.ConnectToNSQDs(nsqdTCPAddrs)
if err != nil {
log.Fatal(err)
}
// 4. connect to lookupd
err = consumer.ConnectToNSQLookupds(lookupdHTTPAddrs)
if err != nil {
log.Fatal(err)
}
consumers = append(consumers, consumer)
下面來看下每個部分的實際程式碼：
func (r *Consumer) AddHandler(handler Handler) {
r.AddConcurrentHandlers(handler, 1)
}
func (r *Consumer) AddConcurrentHandlers(handler Handler, concurrency int) {
if atomic.LoadInt32(&r.connectedFlag) == 1 {
panic("already connected")
}
atomic.AddInt32(&r.runningHandlers, int32(concurrency))
for i := 0; i < concurrency; i++ {
go r.handlerLoop(handler)
}
}
至此handler新增完成，起一個單獨的go routine來等待訊息的到了。
func (r *Consumer) handlerLoop(handler Handler) {
r.log(LogLevelDebug, "starting Handler")
for {
message, ok := <-r.incomingMessages // 有新的訊息的到來
if !ok {
goto exit
}
if r.shouldFailMessage(message, handler) {
message.Finish()
continue
}
err := handler.HandleMessage(message) // 呼叫之前註冊的handler
if err != nil {
r.log(LogLevelError, "Handler returned error (%s) for msg %s", err, message.ID)
if !message.IsAutoResponseDisabled() {
message.Requeue(-1)
}
continue
}
if !message.IsAutoResponseDisabled() {
message.Finish()
}
}
exit:
r.log(LogLevelDebug, "stopping Handler")
if atomic.AddInt32(&r.runningHandlers, -1) == 0 {
r.exit()
}
}
官方是不推薦只部署nqd而不部署lookupd的，我們直接看下lookup的連線過程：
func (r *Consumer) ConnectToNSQLookupd(addr string) error {
...
r.lookupdHTTPAddrs = append(r.lookupdHTTPAddrs, addr)
numLookupd := len(r.lookupdHTTPAddrs)
r.mtx.Unlock()
// if this is the first one, kick off the go loop
if numLookupd == 1 {
r.queryLookupd()
r.wg.Add(1)
go r.lookupdLoop()
}
return nil
}
在queryLookupd中先去查詢lookupd獲取最新的nqd地址，然後connect to nsqd.
func (r *Consumer) lookupdLoop() {
// add some jitter so that multiple consumers discovering the same topic,
// when restarted at the same time, dont all connect at once.
ticker = time.NewTicker(r.config.LookupdPollInterval)
// 每個ticker interval更新nqd的地址資訊
for {
select {
case <-ticker.C:
r.queryLookupd()
case <-r.lookupdRecheckChan:
r.queryLookupd()
case <-r.exitChan:
goto exit
}
}
}
func (r *Consumer) ConnectToNSQD(addr string) error {
// 1. new connection
conn := NewConn(addr, &r.config, &consumerConnDelegate{r})
conn.SetLogger(logger, logLvl,
fmt.Sprintf("%3d [%s/%s] (%%s)", r.id, r.topic, r.channel))
// 2. connection list
_, pendingOk := r.pendingConnections[addr]
_, ok := r.connections[addr]
r.pendingConnections[addr] = conn
if idx := indexOf(addr, r.nsqdTCPAddrs); idx == -1 {
r.nsqdTCPAddrs = append(r.nsqdTCPAddrs, addr)
}
r.log(LogLevelInfo, "(%s) connecting to nsqd", addr)
// 3. new connect
//3.1 go c.readLoop()
//3.2 go c.writeLoop()
resp, err := conn.Connect()
// 4. sub to nsqd
cmd := Subscribe(r.topic, r.channel)
err = conn.WriteCommand(cmd)
}
以上就是客戶端初始化的一個流程，然後就是接受訊息處理了。
->NewConsumer() // 新建一個consumer ->ConnectToNSQLookupds() // 連線到lookupd |->ConnectToNSQLookupd() // 連線到lookupd |->r.queryLookupd() // 查詢lookupd的 |->apiRequestNegotiateV1() // 呼叫lookupd的rest api獲取nsqd訊息 |->ConnectToNSQD() // 連線到具體nsq |->NewConn() // 連線instance |->conn.Connect() // 開始連線 |->c.readLoop() // 與nqd連線read loop |->c.writeLoop() // 與nqd連線write loop |->Subscribe() // consumer傳送SUB command |->lookupdLoop() // 定時查詢lookupd並更新nsqd資訊
注：
[1]. 關於delegate模式參考ofollow,noindex" target="_blank">這裡