帶你入門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 = https://www.cnblogs.com/smartrui/p/&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這些專業的訊息佇列，不過在某些場景下還是夠用的，這個就得根椐自已的情況去取舍了，畢竟，沒有好的架構，只有合適的架構，

標籤：Go

上一篇：channel的關閉的經典場景

下一篇：defer 鏈如何被遍歷