訊息佇列廢話不多說了，直切正題吧：
直接去https://github.com/nsqio/nsq/releases下載編譯好的執行檔案，比如我下載的是nsq-1.2.0.linux-amd64.go1.12.9.tar.gz

直接解壓后(我的重命名了nsq-1.2.0)：bin目錄里就會出現一大堆nsq_…開頭的可執行檔案：

進入bin目錄

cd nsq-1.2.0/bin

啟動nsqlookup

./nsqlookupd &

啟動nsqd

./nsqd --lookupd-tcp-address=127.0.0.1:4160 &

運行 nsqadmin 管理

./nsqadmin --lookupd-http-address=0.0.0.0:4161 --http-address=10.20.23.29:8761&

客戶端寫入訊息

curl -d 'hello world 1' 'http://127.0.0.1:4151/pub?topic=hello word'

佇列訊息寫入檔案

./nsq_to_file --topic=test --output-dir=/tmp/log --lookupd-http-address=127.0.0.1:4161

訪問nsqadmin（以下是本機ip）

http://10.20.23.29:8761/

上面是結束了nsql的常用使用組件，我們再來回顧下上面的概念問題：

Nsq服務端

nsqlookupd

看看官方的原話是怎么說：
nsqlookupd是守護行程負責管理拓撲資訊，客戶端通過查詢 nsqlookupd 來發現指定話題（topic）的生產者，并且 nsqd 節點廣播話題（topic）和通道（channel）資訊

簡單的說nsqlookupd就是中心管理服務，它使用tcp(默認埠4160)管理nsqd服務，使用http(默認埠4161)管理nsqadmin服務，同時為客戶端提供查詢功能

總的來說，nsqlookupd具有以下功能或特性

唯一性，在一個Nsq服務中只有一個nsqlookupd服務，當然也可以在集群中部署多個nsqlookupd，但它們之間是沒有關聯的
去中心化，即使nsqlookupd崩潰，也會不影響正在運行的nsqd服務
充當nsqd和naqadmin資訊互動的中間件
提供一個http查詢服務，給客戶端定時更新nsqd的地址目錄

nsqadmin

官方原話：是一套 WEB UI，用來匯集集群的實時統計，并執行不同的管理任務

總的來說，nsqadmin具有以下功能或特性

提供一個對topic和channel統一管理的操作界面以及各種實時監控資料的展示，界面設計的很簡潔，操作也很簡單
展示所有message的數量，恩…裝X利器
能夠在后臺創建topic和channel，這個應該不常用到
nsqadmin的所有功能都必須依賴于nsqlookupd，nsqadmin只是向nsqlookupd傳遞用戶操作并展示來自nsqlookupd的資料

nsqadmin默認的訪問地址是　http://127.0.0.1:4171/

使用

創建一個channer001主題，并發送一個how are you訊息

curl -d 'how are you' 'http://127.0.0.1:4151/pub?topic=channer001'

nsqd

官方原話：nsqd 是一個守護行程，負責接收，排隊，投遞訊息給客戶端

簡單的說，真正干活的就是這個服務，它主要負責message的收發，佇列的維護，nsqd會默認監聽一個tcp埠(4150)和一個http埠(4151)以及一個可選的https埠

總的來說，nsqd 具有以下功能或特性

對訂閱了同一個topic，同一個channel的消費者使用負載均衡策略（不是輪詢）
只要channel存在，即使沒有該channel的消費者，也會將生產者的message快取到佇列中（注意訊息的過期處理）
保證佇列中的message至少會被消費一次，即使nsqd退出，也會將佇列中的訊息暫存磁盤上(結束行程等意外情況除外)
限定記憶體占用，能夠配置nsqd中每個channel佇列在記憶體中快取的message數量，一旦超出，message將被快取到磁盤中
topic，channel一旦建立，將會一直存在，要及時在管理臺或者用代碼清除無效的topic和channel，避免資源的浪費

這是官方的圖，第一個channel(meteics)因為有多個消費者，所以觸發了負載均衡機制，后面兩個channel由于沒有消費者，所有的message均會被快取在相應的佇列里，直到消費者出現
在這里插入圖片描述

使用Go操作NSQ

官方提供了Go語言版的客戶端：go-nsq，更多客戶端支持請查看CLIENT LIBRARIES，

安裝

go get -u github.com/nsqio/go-nsq

1.生產者

一個簡單的生產者示例代碼如下：

// gin-vue/product.go
package main

import (
    "log"
    "github.com/nsqio/go-nsq"
    "io/ioutil"
    "strconv"
)

var nullLogger = log.New(ioutil.Discard, "", log.LstdFlags)

func sendMessage() {
    config := nsq.NewConfig() // 1. 創建生產者
    producer, err := nsq.NewProducer("127.0.0.1:4150", config)
    if err != nil {
        log.Fatalln("連接失敗: (127.0.0.1:4150)", err)
    }

    errPing := producer.Ping() // 2. 生產者ping
    if errPing != nil {
        log.Fatalln("無法ping通: 127.0.0.1:4150", errPing)
    }

    producer.SetLogger(nullLogger, nsq.LogLevelInfo) // 3. 設定不輸出info級別的日志

    for i := 0; i < 5; i++ { // 4. 生產者發布訊息
        message := "訊息發送測驗 " + strconv.Itoa(i+10000)
        err2 := producer.Publish("one-test", []byte(message)) // 注意one-test　對應消費者consumer.go　保持一致
        if err2 != nil {
            log.Panic("生產者推送訊息失敗!")
        }
    }

    producer.Stop() // 5. 生產者停止執行
}

func main() {
    sendMessage()
}

2.消費者

創建消費者的方式有兩種，一種是通過http請求來發現nsqd生產者和配置的topic，另一種是直接使用tcp請求來連接本地實體，

示例代碼如下所示：

// gin-vue/consumer.go
package main

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

func doConsumerTask() {
    // 1. 創建消費者
    config := nsq.NewConfig()
    q, errNewCsmr := nsq.NewConsumer("one-test", "ch-one-test", config)
    if errNewCsmr != nil {
        fmt.Printf("fail to new consumer!, topic=%s, channel=%s", "one-test", "ch-one-test")
    }

    // 2. 添加處理訊息的方法
    q.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
        log.Printf("message: %v", string(message.Body))
        message.Finish()
        return nil
    }))

    // 3. 通過http請求來發現nsqd生產者和配置的topic（推薦使用這種方式）
    lookupAddr := []string{
        "127.0.0.1:4161",
    }
    err := q.ConnectToNSQLookupds(lookupAddr)
    if err != nil {
        log.Panic("[ConnectToNSQLookupds] Could not find nsqd!")
    }

    // 4. 接收消費者停止通知
    <-q.StopChan

    // 5. 獲取統計結果
    stats := q.Stats()
    fmt.Sprintf("message received %d, finished %d, requeued:%s, connections:%s",
        stats.MessagesReceived, stats.MessagesFinished, stats.MessagesRequeued, stats.Connections)
}

func main() {
    doConsumerTask()
}