NSQ消息隊列怎么在Golang中安裝與使用

NSQ消息隊列怎么在Golang中安裝與使用？很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

前言

NSQ是Go語言編寫的，開源的分布式消息隊列中間件，其設計的目的是用來大規模地處理每天數以十億計級別的消息。NSQ 具有分布式和去中心化拓撲結構，該結構具有無單點故障、故障容錯、高可用性以及能夠保證消息的可靠傳遞的特征，是一個成熟的、已在大規模生成環境下應用的產品。

背景介紹

在服務器最開始的時候，基本上在一臺主機上就能解決大部分問題，所以一般架構設計如下：

但是，突然某一天，來了一個新需求，我們服務器上不只是簡單的儲存一些文本信息，我們需要儲存圖片甚至視頻，顯然直接在一臺主機上再部署一個文件服務是不科學的，因為它會占用大量的流量，這時候我們就要考慮單獨使用一個文件服務器了，所以我們調整一下架構：

通過nginx這個高性能代理服務器，有效的處理不同用戶請求的訴求：

然后接下來某一天，隨著用戶越來越多，你發現服務器的負載越來越高，這個時候，就要考慮去分離業務服務和數據庫服務了，設計再次修改：

接下來又過了一些日子，新的問題又出現了，你們公司的業務越做越大，用戶越來越多，有用戶投訴，收到的響應延遲很大甚至出現了無響應的情況，你檢查了一下服務器，發現每天峰值的時候，有一小部分數據是經常要使用的，頻繁的從數據庫讀出這部分數據，占用了大量的數據庫資源，然后我們的設計又要改了：

把一些常用的數據儲存到緩存中，遵循國際慣例的二八原則（80%的請求讀取20%的數據），這樣一來，數據庫的壓力負擔自然可以減少很多。

當你覺得高枕無憂的時候，突然老板一時興起，我們來搞一個秒殺活動吧，這時候你一定會絞盡腦汁解決不能多買多賣的問題（一瞬間高并發的常見問題），好吧，終于輪到我們消息隊列出場了（當然消息隊列不是唯一的解決方案），我們先把設計貼出來：

好的，所有的請求都先注入消息隊列，然后立刻給予請求響應，因為注入消息隊列實際上向內存寫入數據的過程，所以響應的速度會非常的快，然后后面的業務服務器再根據消息的隊列的內容逐個讀出（相當于異步讀取），所以可以大大削減數據庫的壓力，避免多買多賣的問題。當然后面隨著業務的擴大，還有很多問題要解決，比如使用負載均衡搭建多個業務服務器，使用分布式部署數據庫，搭載高可用架構等，但是今天只是僅僅為了引出消息隊列它的應用場景，所以我們就不往下討論了。

正文

打開 https://nsq.io/deployment/installing.html 下載對應的nsq版本，我下載的是linux最新穩定版

下載解壓之后，在/usr/下建立一個目錄，接著把解壓文件夾/bin/下面的文件全部拷貝進去，最后在/etc/profile添加引用路徑，這樣就可以直接使用命令啟動nsq服務了，我的配置如下

我們先介紹一下幾個必要服務的作用

nsqlookupd：

主要負責服務發現 負責nsqd的心跳、狀態監測，給客戶端、nsqadmin提供nsqd地址與狀態，啟動命令如下：

nsqd：

負責接收消息，存儲隊列和將消息發送給客戶端，nsqd 可以多機器部署，當你使用客戶端向一個topic發送消息時，可以配置多個nsqd地址，消息會隨機的分配到各個nsqd上，nsqd優先把消息存儲到內存channel中，當內存channel滿了之后，則把消息寫到磁盤文件中。他監聽了兩個tcp端口，一個用來服務客戶端，一個用來提供http的接口 ，啟動命令如下：

nsqadmin：

nsqadmin是一個web管理界面 啟動方式如下：

啟動之后，通過 http://10.10.6.147:4171/ 可以訪問這個管理頁面， 默認使用4171端口

我們先來說明一下這個后臺里面的一些內容，因為我們的NSQ所使用的是經典的pub/sub模式（發布/訂閱，典型的生產者/消費者模式），我們可以先發布一個主題到NSQ，然后所有訂閱的服務器就會異步的從這里讀取主題的內容：

Topic(左上角)：發布的主題名字

NSQd Host：Nsq主機服務地址

Channel：消息通道

NSQd Host：Nsq主機服務地址

Depth：消息積壓量

In-flight：已經投遞但是還未消費掉的消息

Deferred：沒有消費掉的延時消息

Messages：服務器啟動之后，總共接收到的消息量

Connections：通道里面客戶端的訂閱數

TimeOut：超時時間內沒有被響應的消息數

Memory + Disk：儲存在內存和硬盤中總共的消息數

----------------------------------------華麗的分割線----------------------------------------

接著，我們講解如何在代碼中發布主題內容，然后通過訂閱某主題去異步讀取消息

先創建一個主題，并且發布100條消息：

package main 
import (
 "github.com/nsqio/go-nsq"
 "fmt"
)
 
var (
 //nsqd的地址，使用了tcp監聽的端口
 tcpNsqdAddrr = "10.10.6.147:4150"
)
 
func main() {
 //初始化配置
 config := nsq.NewConfig()
 for i := 0; i < 100; i++ {
 //創建100個生產者
 tPro, err := nsq.NewProducer(tcpNsqdAddrr, config)
 if err != nil {
 fmt.Println(err)
 }
 //主題
 topic := "Insert"
 //主題內容
 tCommand := "new data!"
 //發布消息
 err = tPro.Publish(topic, []byte(tCommand))
 if err != nil {
 fmt.Println(err)
 }
 } 
}

接下來我們看看admin的顯示內容：

我們可以看到Nsqd接收到了100條信息，100條信息都儲存在內存中，沒有被消化。

現在沒有任何服務訂閱了我們的主題，所以主題的消息都沒有被消化，那我們創建一個消費者去訂閱我們的主題：

package main 
import (
 "github.com/nsqio/go-nsq"
 "fmt"
 "sync"
 "time"
)
 
var (
 //nsqd的地址，使用了tcp監聽的端口
 tcpNsqdAddrr = "10.10.6.147:4150"
)
 
//聲明一個結構體，實現HandleMessage接口方法（根據文檔的要求）
type NsqHandler struct {
 //消息數
 msqCount int64
 //標識ID
 nsqHandlerID string
}
 
//實現HandleMessage方法
//message是接收到的消息
func (s *NsqHandler) HandleMessage(message *nsq.Message) error {
 //沒收到一條消息+1
 s.msqCount++
 //打印輸出信息和ID
 fmt.Println(s.msqCount,s.nsqHandlerID)
 //打印消息的一些基本信息
 fmt.Printf("msg.Timestamp=%v, msg.nsqaddress=%s,msg.body=%s \n", time.Unix(0 , message.Timestamp).Format("2006-01-02 03:04:05") , message.NSQDAddress, string(message.Body))
 return nil
} 
func main() {
 
 //初始化配置
 config := nsq.NewConfig()
 //創造消費者，參數一時訂閱的主題，參數二是使用的通道
 com, err := nsq.NewConsumer("Insert", "channel1", config)
 if err != nil {
 fmt.Println(err)
 }
 //添加處理回調
 com.AddHandler(&NsqHandler{nsqHandlerID:"One"})
 //連接對應的nsqd
 err = com.ConnectToNSQD(tcpNsqdAddrr)
 if err != nil {
 fmt.Println(err)
 }
 
 //只是為了不結束此進程，這里沒有意義
 var wg = &sync.WaitGroup{}
 wg.Add(1)
 wg.Wait() 
 /*
 result: 
 msg.Timestamp=2018-11-02 04:37:18, msg.nsqaddress=10.10.6.147:4150,msg.body=new data! 
 98 One
 msg.Timestamp=2018-11-02 04:37:18, msg.nsqaddress=10.10.6.147:4150,msg.body=new data! 
 99 One
 msg.Timestamp=2018-11-02 04:37:18, msg.nsqaddress=10.10.6.147:4150,msg.body=new data! 
 100 One
 msg.Timestamp=2018-11-02 04:37:18, msg.nsqaddress=10.10.6.147:4150,msg.body=new data! 
 */ 
}

這里可以看到，之前擠壓的100條信息，都被我們的訂閱者消化掉了，也就是讀取了

所以我們的訂閱者（可以有多個）如果提前訂閱主題的話，只要對應的主題有發布新內容，就可以馬上異步讀取。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。