[[431096]]

01 

介紹

NSQ 是 Golang 語言編寫的實時分布式消息傳遞平臺（也可以理解為消息隊列），它主要由三個守護進程組成，分別是 nsqd 、 nsqlookupd 和  nsqadmin 。其中  nsqd 是核心組成部分，它負責處理客戶端的請求，比如生產、排序和消費消息等； nsqlookupd 負責管理集群拓撲信息和提供一個最終一致性的發現服務， nsqadmin 是一個 web 界面的管理平臺，可以用于實時查看集群信息和執行其他管理操作。單個  nsqd 可以含有很多 topic，每個 topic 可以含有很多 channel。

NSQ 支持跨平臺和多語言客戶端。使用 Mac 的讀者朋友們可以使用 brew 方便的安裝 NSQ。本文我們主要介紹 NSQ 官方提供的 golang 客戶端 go-nsq。關于 NSQ 的更多內容，感興趣的讀者朋友們可以查閱官方文檔，限于篇幅，本文不再贅述。

使用 go-nsq 操作 NSQ，需要安裝  go-nsq 庫，它提供了許多用于操作 NSQ 的函數和方法。

安裝方式：

// Mac 安裝 nsq 
brew install nsq 
// 安裝 go-nsq 
go get -u github.com/nsqio/go-nsq 

02 

生產者

go-nsq 包中的 Producer 類型，用于向 NSQ 發送消息。首先，需要調用函數 NewProducer 創建一個 Producer 實例，接收參數是 string 類型的 addr 和指針類型的 nsq.Config（NSQ 配置信息），返回結果是一個 Producer 實例的地址和 error。

需要注意的是，必須調用函數 NewConfig 返回一個指針類型的 nsq.Config，它包含默認配置信息。可以通過調用該實例的 Set 方法設置配置信息，并且必須在用于傳參之前設置，否則設置的配置信息將不會生效。

Producer 包含很多方法，本文主要介紹四個方法，分別是 Ping、String、Publish 和 Stop。其中 Ping 方法用于檢測 Producer 是否連接成功它配置的 nsqd ；String 方法返回 Producer 連接的  nsqd 的地址；Publish 方法用于同步發送消息到指定的 topic；Stop 方法用于優雅地停止 Producer。

示例代碼：

// 默認配置信息 
config := nsq.NewConfig() 
 
// 創建生產者 
producer, err := nsq.NewProducer("127.0.0.1:4150", config) 
if err != nil { 
 log.Fatal(err) 
} 
// 驗證生成者連接是否成功 
err = producer.Ping() 
if err != nil { 
 log.Fatal(err) 
} 
// 返回生產者地址 
producerAddr := producer.String() 
log.Printf("producerAddr:%v", producerAddr) 
messageBody := []byte("hello") 
topicName := "topic" 
// 同步發送消息到指定 topic 
err = producer.Publish(topicName, messageBody) 
if err != nil { 
 log.Fatal(err) 
} 
producer.Stop() 

運行結果：

2021/10/23 18:58:23 INF    1 (127.0.0.1:4150) connecting to nsqd 
2021/10/23 18:58:23 producerAddr:127.0.0.1:4150 
2021/10/23 18:58:23 INF    1 stopping 
2021/10/23 18:58:23 INF    1 exiting router 

03 

消費者

go-nsq 包中的 Consumer 類型，用于從 NSQ 消費消息。首先，需要調用函數 NewConsumer 創建一個 Consumer 實例，接收參數是 string 類型的 topic 和 channel，指針類型的 nsq.Config（NSQ 配置信息），返回結果是一個 Consumer 實例的地址和 error。

需要注意的是，必須調用函數 NewConfig 返回一個指針類型的 nsq.Config，它包含默認配置信息。可以通過調用該實例的 Set 方法設置配置信息，并且必須在用于傳參之前設置，否則設置的配置信息將不會生效。

Consumer 包含很多方法，本文主要介紹三個方法，分別是 Stats、AddHandler 和 ConnectToNSQD。其中 Stats 方法用于檢索 Consumer 的當前連接和消息統計信息；AddHandler 用于為 Consumer 消費的消息設置處理函數，每個處理函數都獨立運行在一個 goroutine 中，如果需要啟動多個 goroutine 運行處理函數，可以多次調用 AddHandler；ConnectToNSQD 用于連接配置的 nsqd 。

示例代碼：

config := nsq.NewConfig() 
 // 創建 Consumer 
 consumer, err := nsq.NewConsumer("topic", "channel", config) 
 if err != nil { 
  log.Fatal(err) 
 } 
 consumerStats := consumer.Stats() 
 log.Printf("consumerStats:%+v", consumerStats) 
 // 給 Consumer 添加處理器，可添加多個，每個 Handler 都運行在單獨的 goroutine 中 
 consumer.AddHandler(&myMessageHandler{}) 
 // 連接 nsqd 
 err = consumer.ConnectToNSQD("127.0.0.1:4150") 
 if err != nil { 
  log.Fatal(err) 
 } 
 <-consumer.StopChan 

運行結果：

2021/10/23 18:59:30 consumerStats:&{MessagesReceived:0 MessagesFinished:0 MessagesRequeued:0 Connections:0} 
2021/10/23 18:59:30 INF    1 [topic/channel] (127.0.0.1:4150) connecting to nsqd 
2021/10/23 18:59:30 hello 

04 

總結

本文主要介紹 Golang 語言編寫的實時分布式消息平臺 NSQ 的 golang 客戶端 go-nsq。它是 NSQ 官方提供的 NSQ Golang 客戶端。并且 分別介紹了 Producer 和 Consumer 的簡單使用方法，大家可以根據自己的業務需求，使用 go-nsq 靈活運用 NSQ。關于 go-nsq 的更多內容，感興趣的讀者朋友們可以閱讀官方文檔。