1. 概述
NSQ是一個基于Go語言的分布式實時消息平臺舶治，它基于MIT開源協(xié)議發(fā)布冬殃，由bitly公司開源出來的一款簡單易用的消息中間件相關(guān)描述如下:
NSQ是一個實時分布式消息傳遞平臺，旨在大規(guī)模運行较坛，每天處理數(shù)十億條消息捐迫。 它促進了沒有單點故障的分布式和分散式拓撲卢厂，從而實現(xiàn)了容錯能力和高可用性，并提供了可靠的消息傳遞保證河劝。 查看功能和保證壁榕。 從操作上講，NSQ易于配置和部署（所有參數(shù)均在命令行上指定赎瞎，并且編譯的二進制文件不具有運行時相關(guān)性）牌里。 為了獲得最大的靈活性，它與數(shù)據(jù)格式無關(guān)（消息可以是JSON务甥，MsgPack牡辽，協(xié)議緩沖區(qū)或其他任何東西）。 官方提供了Go和Python庫（以及許多其他客戶端庫）敞临，并且态辛，如果您有興趣構(gòu)建自己的庫，則有一個協(xié)議規(guī)范挺尿。NSQ的特點
- 支持水平橫向拓展(無縫添加更多節(jié)點到集群中)
- 部署配置容易,自帶集群管理界面(nsqadmin)
- 提倡分布式拓撲,減少單點故障,提高容錯
- 低延遲的消息傳遞
- 可靠的消交付保障保障
- 默認中消息都在內(nèi)存中, nsq 內(nèi)部機制保證在程序關(guān)閉時將隊列中的數(shù)據(jù)持久化到硬盤奏黑，重啟后就會恢復(fù)。
- 消息最少被投遞一次
比較知名和常用的消息處理系統(tǒng)還有
2. 基礎(chǔ)應(yīng)用場景
我們知道一般的消息隊列(Message Queue) 常用的場景有
系統(tǒng)解耦異步處理流量削峰消息通信等
3. 相關(guān)文檔
- 項目地址 : https://github.com/nsqio/nsq
- 項目文檔 英文: https://nsq.io/overview/design.html
- 下載地址: https://nsq.io/deployment/installing.html
- 客戶端下載地址 : https://nsq.io/clients/client_libraries.html
4.安裝操作
根據(jù)自己的操作平臺下載解壓即可
- 根據(jù)自己的操作系統(tǒng)下載對應(yīng)的壓縮包文件
- 解壓壓縮文件
- 進入解壓后
bin目錄中在
bin目錄中我們能看到如下文件-rwxr-xr-x 1 captain 197121 5515776 8月 28 13:46 nsq_stat.exe* -rwxr-xr-x 1 captain 197121 5823488 8月 28 13:46 nsq_tail.exe* -rwxr-xr-x 1 captain 197121 5997568 8月 28 13:46 nsq_to_file.exe* -rwxr-xr-x 1 captain 197121 5923840 8月 28 13:46 nsq_to_http.exe* -rwxr-xr-x 1 captain 197121 5903872 8月 28 13:46 nsq_to_nsq.exe* -rwxr-xr-x 1 captain 197121 8787968 8月 28 13:46 nsqadmin.exe* -rwxr-xr-x 1 captain 197121 9108992 8月 28 13:46 nsqd.exe* -rwxr-xr-x 1 captain 197121 8384000 8月 28 13:46 nsqlookupd.exe* -rwxr-xr-x 1 captain 197121 5639680 8月 28 13:46 to_nsq.exe*
5. NSQ服務(wù)端基礎(chǔ)組件介紹
5.1 nsqd
nsqd是一個守護進程負責(zé)
接收,排隊,消息傳遞到客戶端票髓。 它可以獨立運行攀涵，但通常由nsqlookupd實例的群集中配置（在這種情況下，它將能聲明topics和發(fā)現(xiàn)channel）洽沟。 它偵聽兩個TCP端口以故，一個偵聽客戶端，另一個偵聽HTTP API裆操。 它可以選擇在第三個端口上偵聽HTTPS怒详。
5.2 nsqlookupd
nsqlookupd是管理拓撲信息的守護程序炉媒。 客戶端查詢nsqlookupd以發(fā)現(xiàn)特定topic的nsqd生產(chǎn)者和nsqd節(jié)點廣播topic和channel信息。
有兩個接口：nsqd用于廣播的TCP接口
客戶端(nsqadmin)執(zhí)行發(fā)現(xiàn)和管理操作的HTTP接口
5.3 nsqadmin
nsqadmin是一套 WEB管理界面,用來匯集集群的實時統(tǒng)計昆烁，并執(zhí)行不同的管理任務(wù)吊骤。
重點提示:
NSQ還有許多功能組件,我們只介紹這三個(
nsqdnsqlookupdnsqadmin)最常用和主要的NSQ的所有組件都可以通過參數(shù)
-- help查看相關(guān)配置
nsqd和nsqlookupd都有對應(yīng)的http API ,需要使用的時候查看文檔即可
6.操作NSQ
6.1 安裝客戶端
根據(jù)不同的開發(fā)語言選擇不同的客戶端
我們是使用Golang操作所以采用NSQ的官方提供客戶端 go-nsq
go get -u github.com/nsqio/go-nsq
6.1 單機啟動nsqd
默認啟動的
nsqd監(jiān)聽 HTTP對應(yīng)的4151端口和TCP對應(yīng)的4150端口
$ ./nsqd
[nsqd] 2019/11/10 13:41:29.575014 INFO: nsqd v1.2.0 (built w/go1.12.9)
[nsqd] 2019/11/10 13:41:29.593002 INFO: ID: 825
[nsqd] 2019/11/10 13:41:29.597000 INFO: TOPIC(topic_demo): created
[nsqd] 2019/11/10 13:41:29.599998 INFO: TOPIC(topic_demo): new channel(aa)
[nsqd] 2019/11/10 13:41:29.599998 INFO: NSQ: persisting topic/channel metadata to nsqd.dat
[nsqd] 2019/11/10 13:41:29.644973 INFO: HTTP: listening on [::]:4151
[nsqd] 2019/11/10 13:41:29.644973 INFO: TCP: listening on [::]:4150
我們同樣可以指定端口
$ ./nsqd -http-address="0.0.0.0:8080" -tcp-address="0.0.0.0:8081"
[nsqd] 2019/11/10 14:05:40.726849 INFO: nsqd v1.2.0 (built w/go1.12.9)
[nsqd] 2019/11/10 14:05:40.745838 INFO: ID: 825
[nsqd] 2019/11/10 14:05:40.747836 INFO: NSQ: persisting topic/channel metadata to nsqd.dat
[nsqd] 2019/11/10 14:05:40.788814 INFO: TCP: listening on [::]:8081
[nsqd] 2019/11/10 14:05:40.788814 INFO: HTTP: listening on [::]:8080
這樣我們就啟動了一個nsqd 的實例
在NSQ中有兩個非常重要的概念
topic和Channel我們看一下文檔中的描述:
每個nsqd實例旨在一次處理多個數(shù)據(jù)流。這些數(shù)據(jù)流稱為
“topics”静尼，一個topic具有1個或多個“channels”白粉。每個channel都會收到topic所有消息的副本，實際上下游的服務(wù)是通過對應(yīng)的channel來消費topic消息鼠渺。
topic和channel不是預(yù)先配置的鸭巴。topic在首次使用時創(chuàng)建，方法是將其發(fā)布到指定topic拦盹，或者訂閱指定topic上的channel鹃祖。channel是通過訂閱指定的channel在第一次使用時創(chuàng)建的。
topic和channel都相互獨立地緩沖數(shù)據(jù)普舆，防止緩慢的消費者導(dǎo)致其他chennel的積壓（同樣適用于topic級別）恬口。
channel可以并且通常會連接多個客戶端。假設(shè)所有連接的客戶端都處于準(zhǔn)備接收消息的狀態(tài)沼侣，則每條消息將被傳遞到隨機客戶端祖能。例如：topic-channel.gif
總而言之，消息是從
topic->channel多播的（每個channel都接收該topic的所有消息的副本）华临，但從channel->消息消費者均勻分發(fā)（每個消費者都接收該頻道的一部分消息）芯杀。
6.1.1 單NSQ的使用
編寫一個消息生產(chǎn)者
nsq_single_product.go
package main
import (
"fmt"
"github.com/nsqio/go-nsq"
"time"
)
func main() {
nsqAddr := "127.0.0.1:8081"
conf :=nsq.NewConfig()
p ,err := nsq.NewProducer(nsqAddr,conf)
if err != nil {
fmt.Println(err)
return
}
for {
message := "message :"+ time.Now().Format("2006-01-02 15:04:05")
fmt.Println(message)
// 發(fā)送消息
p.Publish("topic-demo1",[]byte(message))
time.Sleep(time.Second)
}
}
編寫一個消息消費者
nsq_single_consumer.go
package main
import (
"fmt"
"github.com/nsqio/go-nsq"
)
type NewHandler struct{}
func (m *NewHandler) HandleMessage(msg *nsq.Message) (err error) {
addr := msg.NSQDAddress
message := string(msg.Body)
fmt.Println(addr, message)
return
}
func MyConsumers(topic, channel, addr string) {
conf := nsq.NewConfig()
new_consumer, err := nsq.NewConsumer(topic, channel, conf)
if err != nil {
}
// 接收消息
new_handler := &NewHandler{}
new_consumer.AddHandler(new_handler)
err = new_consumer.ConnectToNSQD(addr)
if err != nil {
}
}
func main() {
addr := "127.0.0.1:8081"
go MyConsumers("topic-demo1", "channel-aa", addr)
// 模擬多個從多個channel去消息
go MyConsumers("topic-demo1", "channel-bb", addr)
select {}
}
6.1.2 通過nsqadmin查看
啟動
nsqadmin
nsqadmin的web界面默認監(jiān)聽了 4171端口
$ ./nsqadmin --nsqd-http-address="127.0.0.1:8080"
[nsqadmin] 2019/11/10 16:06:15.842033 INFO: nsqadmin v1.2.0 (built w/go1.12.9)
[nsqadmin] 2019/11/10 16:06:15.858026 INFO: HTTP: listening on [::]:4171
我們在地址欄中輸如
http://127.0.0.1:4171/就能看看管理界面
6.1.3 NSQ的單點結(jié)構(gòu)
6.3 NSQ集群
6.3.1 啟動NSQ各組件
構(gòu)建一個NSQ的基礎(chǔ)拓撲結(jié)構(gòu)
我們可以簡單的說
nsqlookupd是用來管理nsqd實例節(jié)點的
第一步
啟動nsqlookupd啟動的
nsqlookupd采用了默認配置 通過參數(shù)--help查看配置項
$ ./nsqlookupd
[nsqlookupd] 2019/11/10 16:40:55.968588 INFO: nsqlookupd v1.2.0 (built w/go1.12.9)
[nsqlookupd] 2019/11/10 16:40:55.983580 INFO: HTTP: listening on [::]:4161
[nsqlookupd] 2019/11/10 16:40:55.984579 INFO: TCP: listening on [::]:4160
第二步
添加
nsqd實例與前面的啟動不同,需要帶上參數(shù)
-lookupd-tcp-address
添加第一個實例
./nsqd -http-address="0.0.0.0:8080" -tcp-address="0.0.0.0:8081" -lookupd-tcp-address="127.0.0.1:4160"
添加第二個實例
./nsqd -http-address="0.0.0.0:8090" -tcp-address="0.0.0.0:8091" -lookupd-tcp-address="127.0.0.1:4160"
第三步
啟動nsqadmin
與前面的也不同了需要帶上參數(shù)
-lookupd-http-address
$ ./nsqadmin -lookupd-http-address="127.0.0.1:4161"
在瀏覽器中訪問
nsqadmin
6.3.2 NSQ的拓撲結(jié)構(gòu)
- 在集群模式中,消息生產(chǎn)方發(fā)送消息給任意一個
nsqd實例都不影響- 消息的消費者需要通過nsqlookupd 查詢nsqd的地址后才能獲取消息
- 增加
nsqd節(jié)點完全不影響其他的節(jié)點
6.3.3 Golang使用NSQ代碼示例
消息生產(chǎn)者
nsq_cluster_product.go
package main
import (
"bufio"
"fmt"
"github.com/nsqio/go-nsq"
"log"
"os"
"strings"
)
var pro *nsq.Producer
func NewPro(addr string) (err error) {
conf := nsq.NewConfig()
pro, err = nsq.NewProducer(addr, conf)
if err != nil {
log.Println(err)
return err
}
return nil
}
func main() {
nsqAddr := "127.0.0.1:8091"
err := NewPro(nsqAddr)
if err != nil {
fmt.Println(err)
return
}else{
fmt.Println("connect 127.0.0.1:8091 success")
}
// 讀取標(biāo)準(zhǔn)輸入
reader := bufio.NewReader(os.Stdin)
for {
// 讀取所有內(nèi)容直到遇見回車(\n)
data, err := reader.ReadString('\n')
if err != nil {
fmt.Println("read data from stdin is field : ", err)
continue
}
// 當(dāng)輸入q的時候退出
data = strings.TrimSpace(data)
if strings.ToUpper(data) == "Q" {
break
}
err = pro.Publish("topic-demo1", []byte(data))
if err != nil {
fmt.Println("nsq publish is field ", err)
continue
}
}
fmt.Println("exit !")
}
消息消費者
nsq_cluster_consumer.go
package main
import (
"fmt"
"github.com/nsqio/go-nsq"
)
type Handler struct{}
func (m *Handler) HandleMessage(msg *nsq.Message) (err error) {
addr := msg.NSQDAddress
message := string(msg.Body)
fmt.Println(addr, message)
return
}
func NewConsumers(t string, c string, addr string) error {
conf := nsq.NewConfig()
nc, err := nsq.NewConsumer(t, c, conf)
if err != nil {
fmt.Println("create consumer failed err ", err)
return err
}
consumer := &Handler{}
nc.AddHandler(consumer)
// 連接nsqlookupd
if err:= nc.ConnectToNSQLookupd(addr);err!=nil{
fmt.Println("connect nsqlookupd failed ", err)
return err
}
return nil
}
func main() {
// 這是nsqlookupd的地址
addr := "127.0.0.1:4161"
err := NewConsumers("topic-demo1", "channel-aa", addr)
if err != nil {
fmt.Println("new nsq consumer failed", err)
return
}
select {}
}
