帶你入門Go的消息隊列NSQ

以前看到過NSQ這個東西，也一直沒去看。今天剛好有時間就搭建了下，簡單嘗試了下這個Go語言下的消息隊列NSQ，我這里簡要記錄下。

其實，NSQ國內用的是比較少的，我這里也是算了解這么個東西吧 ，稍微看下源碼，學到東西而已。

NSQ簡介

NSQ是一個基於Go語言的分布式實時消息平台, 它具有分布式、去中心化的拓撲結構，支持無限水平擴展。無單點故障、故障容錯、高可用性以及能夠保證消息的可靠傳遞的特征。另外，NSQ非常容易配置和部署, 且支持眾多的消息協議。支持多種客戶端，協議簡單，如果有興趣可參照協議自已實現一個也可以。

NSQ的幾個組件

1. nsqd：一個負責接收、排隊、轉發消息到客戶端的守護進程
2. nsqlookupd：管理拓撲信息, 用於收集nsqd上報的topic和channel,並提供最終一致性的發現服務的守護進程
3. nsqadmin：一套Web用戶界面，可實時查看集群的統計數據和執行相應的管理任務
4. utilities：基礎功能、數據流處理工具，如nsq_stat、nsq_tail、nsq_to_file、nsq_to_http、nsq_to_nsq、to_nsq

相關網址

NSQ官網：NSQ官網

GitHub: Github

NSQ官方客戶端：NSQ官方客戶端

NSQ文檔：NSQ文檔

NSQ協議：NSQ協議

NSQ安裝

NSQ的安裝方式有好幾種，可以通過二進制、源碼、Docker、Brew等方式安裝

二進制安裝，可以到安裝地址 上面下載對應平台的Release包，然后解壓就行了。

如果是Mac電腦，直接用Brew安裝

brew install nsq

如果是Docker的安裝，就參照下上面那個網址吧，按照步驟操作就行了，我沒有用的這個安裝方式。

我是用的源碼的安裝方式，因為二進制的那個放在S3上面，下起來好慢，於是直接把Github的源代碼下載來，這里也有一個好處就是可以看源碼來跟蹤學習。還方便些。

下載后的目錄結構如下所示：

NSQ 運行

如果用源碼運行，而不是Make后將可執行文件放到bin目錄這種，那么下載后解決完所有的依賴包后，cd 進入到 nsqio/nsq/apps/nsqd目錄后，可以執行 go run ./ 或 go run main.go options.go 否則會報如下錯誤

nsqio/nsq/apps/nsqd/main.go:44:13: undefined: nsqdFlagSet
nsqio/nsq/apps/nsqd/main.go:54:10: undefined: config

其實進入到apps目錄執行，最終還是會到 nsqio/nsq/nsqd這個下面去執行業務處理代碼的，apps這里僅僅是用go-srv這個包進行了一層服務包裝而已，變成守護和一些入口參數等。

$ go run ./
[nsqd] 2020/03/22 00:55:27.597911 INFO: nsqd v1.2.1-alpha (built w/go1.11.2)
[nsqd] 2020/03/22 00:55:27.597980 INFO: ID: 809
[nsqd] 2020/03/22 00:55:27.598396 INFO: TOPIC(test): created
[nsqd] 2020/03/22 00:55:27.598449 INFO: TOPIC(test): new channel(test)
[nsqd] 2020/03/22 00:55:27.598535 INFO: TOPIC(test): new channel(lc)
[nsqd] 2020/03/22 00:55:27.598545 INFO: NSQ: persisting topic/channel metadata to nsqd.dat
[nsqd] 2020/03/22 00:55:27.599714 INFO: TCP: listening on [::]:4150
[nsqd] 2020/03/22 00:55:27.599806 INFO: HTTP: listening on [::]:4151

看到上面的提示，表示啟動成功了，它會分別開放TCP和HTTP的端口，4150,4151可以通過配置或flag參數的方式更改， 同時它也支持TLS/SSL.

HTTP測試

啟動nsqd后，可以用http來測試發送一條消息,可使用CURL來操作。

$ curl -d '這是一條測試消息' 'http://127.0.0.1:4151/pub?topic=test&channel=lc'
OK

NSQ消息模式

我們知道消息一般有推和拉模式，NSQ的消息模式為推的方式，這種模式可以保證消息的及時性，當有消息時可以及時推送出去。但是要根椐客戶端的消耗能力和節奏去控制，NSQ是通過更改RDY的值來實現的。當沒有消息時為0, 服務端推送消息后，客戶端比如調用 updateRDY（）這個方法改成3, 那么服務端推送時，就會根椐這個值做流控了。

發送消息是通過連接的TCP發出去的，先發到Topic下面，再轉到Channel下面，最后從通道 memoryMsgChan 中取出msg，然后發出。

github.com/nsqio/nsq/nsqd/protocol_v2.go

func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
	var err error
	var memoryMsgChan chan *Message
	var backendMsgChan <-chan []byte
	var subChannel *Channel
	// NOTE: `flusherChan` is used to bound message latency for
	// the pathological case of a channel on a low volume topic
	// with >1 clients having >1 RDY counts
	var flusherChan <-chan time.Time
	var sampleRate int32

	subEventChan := client.SubEventChan
	identifyEventChan := client.IdentifyEventChan
	outputBufferTicker := time.NewTicker(client.OutputBufferTimeout)
	heartbeatTicker := time.NewTicker(client.HeartbeatInterval)
	heartbeatChan := heartbeatTicker.C
	msgTimeout := client.MsgTimeout
	
	...
    ...
	case msg := <-memoryMsgChan:
			if sampleRate > 0 && rand.Int31n(100) > sampleRate {
				continue
			}
			msg.Attempts++

			subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
			client.SendingMessage()
			err = p.SendMessage(client, msg)
			if err != nil {
				goto exit
			}
			flushed = false
		case <-client.ExitChan:
			goto exit
		}

NSQ還支持延時消息的發送，比如訂單在30分鍾未支付做無效處理等場景，延時使用的是heap包的優級先隊列，實現了里面的一些方法。通過判斷當前時間和延時時間做對比，然后從延時隊列里面彈出消息再發送到channel中，后續流程和普通消息一樣，我看網上有 人碰到過說延時消息會有並發問題，最后還用的Redis的ZSET實現的，所以不確定這個延時的靠不靠譜，要求不高的倒是可以試試。

curl -d '這是一條延遲消息' 'http://127.0.0.1:4151/pub?topic=test&channel=lc&defer=3000'

defer參數，單位：毫秒

NSQ消費

消費消息時，channel類似於kafka里面的消費組的概念，比如同一個channel。那么只會被一個實例消費，不會多個實例都能消費到那條消息，所以可用於消息的負載均衡, 我看到網上有人有疑惑就是他指定topic，然后再用不同的channel去消費，說怎么能收到其它channel的消息，不能直接過濾消息，其實channel不是用來過濾的。

NSQ發送的消息可以確保至少被一個消費者消費，它的消費級別為至少消費一次，為了確保消息消費，如果客戶端超時、重新放入隊列或重連等，重復消費是不可避免的，所以客戶端業務流程一定要做消息的冪等處理。

客戶端回復FIN 或者 REQ 表示成功或者重發。如果客戶端未能及時發送，則NSQ將重復發送消息給該客戶端。

另外，NSQ不像 Kafka，我們是能到消息的有序的，但NSQ不行，客戶端收到的消費為無序的。雖然每條消息有一個時間戳，但如果對順序有要求的，那就要注意了。所以，NSQ更適合處理數據量大但是彼此間沒有順序關系的消息。

NSQ的Go客戶端

NSQ是支持多種形式的客戶端的，像HTTP或客戶端庫來操作，而且官方其實還建議使用HTTP的方式，HTTP的方式，直接發GET或POST請求就行了。

這里Go的話，可使用go-nsq這個庫，地址為：go-nsq :go-nsq

go get https://github.com/nsqio/go-nsq

發送消息

package main

import (
	"errors"
	"fmt"
	"github.com/nsqio/go-nsq"
	"time"
)




func main() {
	var (
		p1 *producer
		p2 *producer
	)
	p1 = &producer{}
	p2 = &producer{}
	p1.producer,_ = InitProducer("127.0.0.1:4150")
	p2.producer,_ = InitProducer("127.0.0.1:4150")

	defer p1.producer.Stop()
	defer p2.producer.Stop()

	//p1.publish("test","hello!!!")
	p1.deferredPublish("test", 10 * time.Second,"這是一條延遲消息?")
	fmt.Println("done")
}


type producer struct {
	producer *nsq.Producer
}

func(p *producer) publish(topic string,message string)  (err error){
	if message == "" {
		return errors.New("message is empty")
	}
	if err = p.producer.Publish(topic,[]byte(message)); err != nil {
		fmt.Println(err)
		return  err
	}
	return nil
}
// 延遲消息
func(p *producer) deferredPublish(topic string,delay time.Duration, message string)  (err error){
	if message == "" {
		return errors.New("message is empty")
	}
	if err = p.producer.DeferredPublish(topic,delay, []byte(message)); err != nil {
		fmt.Println(err)
		return  err
	}
	return nil
}

func InitProducer(addr string) (p *nsq.Producer,err error){
	var (
		config *nsq.Config
	)
	config = nsq.NewConfig()
	if p, err  = nsq.NewProducer(addr, config); err != nil {
		return  nil, err
	}
	return p, nil
}

消費消息

package main

import (
	"encoding/json"
	"fmt"
	"github.com/nsqio/go-nsq"
)

//nsqio消費測試

type MyTestHandler struct {
	q *nsq.Consumer
	messageReceive int
}

func (h *MyTestHandler) HandleMessage(message *nsq.Message) error {
	type Data struct {
	}

	var (
		data *Data
		err error
	)
	data = &Data{}
	if err = json.Unmarshal(message.Body, data) ;err != nil {
		fmt.Printf("Id:%s, Msg:%s \n", message.ID, string(message.Body))
		err = nil
	}
	message.Finish()
	return nil
}
func initConsuemr(topic string, channel string) {
	var (
		config *nsq.Config
		h *MyTestHandler
		err  error
	)
	h = &MyTestHandler{
	}

	config = nsq.NewConfig()

	if h.q, err = nsq.NewConsumer(topic, channel, config); err != nil {
		fmt.Println(err)
		return
	}

	h.q.AddHandler(h)
	if err = h.q.ConnectToNSQD("127.0.0.1:4150"); err != nil {
		fmt.Println(err)
	}

	//<-h.q.StopChan
	fmt.Println("stop")
	return

}

func main() {
	initConsuemr("test","test")
	initConsuemr("test","lc")
	select{}
}

總的來說，NSQ的消費是有保障的，能保證消息的可靠性。可用多個 nsqd和nsqlookupd做分布式集群等，使用Go的channel能夠高並發消費，高吞吐量，而且，部署方面也簡單。
不過，給我的感覺還是不如Kafka和RocketMQ這些專業的消息隊列，不過在某些場景下還是夠用的。這個就得根椐自已的情況去取舍了，畢竟，沒有好的架構，只有合適的架構。