帶你入門Go的消息隊列NSQ

来源:https://www.cnblogs.com/smartrui/archive/2020/03/23/12549562.html
-Advertisement-
Play Games

以前看到過NSQ這個東西，也一直沒去看。今天剛好有時間就搭建了下，簡單嘗試了下這個Go語言下的消息隊列NSQ，我這裡簡要記錄下。 其實，NSQ國內用的是比較少的，我這裡也是算瞭解這麼個東西吧 ，稍微看下源碼，學到東西而已。 NSQ簡介 NSQ是一個基於Go語言的分散式實時消息平臺, 它具有分散式、去 ...


以前看到過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 = &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這些專業的消息隊列，不過在某些場景下還是夠用的。這個就得根椐自已的情況去取捨了，畢竟，沒有好的架構，只有合適的架構。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • PHP設計模式—工廠模式之工廠方法模式
    定義： 工廠方法模式：定義一個用於創建對象的介面，讓子類決定實例化哪一個類。工廠方法使一個類的實例化延遲到其子類。 代碼實例： 這裡還是接著上篇簡單工廠模式的那個例子，改成工廠方法模式。 1、創建一個抽象類 ApptEncoder /** * 抽象類 * Class ApptEncoder */ a ...
  • PHP設計模式—工廠模式之簡單工廠模式
    定義： 簡單工廠模式：將調用者和創建者分離，實現解耦，調用者直接向工廠請求，通過工廠去選擇需要實例化的對象，用一個單獨的類來做這個創建實例的過程。 代碼實例： 問題：假設一個關於個人事務管理的項目中有許多類型的對象，其中一個是 Appointment 對象，現在需要通過一種稱為 BloggsCal ...
  • Spring Boot的面試題
    Spring Boot 是微服務中最好的 Java 框架. 我們建議你能夠成為一名 Spring Boot 的專家。本文精選了三十五個常見的Spring Boot知識點，祝你一臂之力！ 問題一 Spring Boot、Spring MVC 和 Spring 有什麼區別？ 1、Spring Sprin ...
  • Harbor鏡像倉庫搭建
    一、Harbor安裝環境要求 硬體環境： 1）CPU 2核以上，最好4核以上 2）記憶體4G以上，最好8G以上 3）硬碟空間至少40G，最好160G以上 軟體環境： 1）docker v17.06以上版本 2）docker compose v1.18.0以上版本 3）Openssl 更新到最新版（一般 ...
  • 爬蟲之selenium
    selenium基本操作 概念：基於瀏覽器自動化的模塊 ：基於手機自動化的模塊的應用 環境的安裝 跟爬蟲之間的關聯？ 可以實現模擬登陸 便捷的捕獲動態載入數據（可見即可得） 基本操作 導包： （web瀏覽器，driver驅動） 必須提供對應瀏覽器的驅動程式（谷歌，火狐...） "谷歌瀏覽器驅動下載地 ...
  • LeetCode 876. 鏈表的中間結點
    我的LeetCode刷題源碼[GitHub]：https://github.com/izhoujie/Algorithmcii LeetCode 876. 鏈表的中間結點 題目 給定一個帶有頭結點 head 的非空單鏈表，返回鏈表的中間結點。 如果有兩個中間結點，則返回第二個中間結點。 示例 1： ...
  • JSP+MySQL+Java開發ssh網上預約預約掛號系統源碼
    JSP+MySQL+Java開發ssh網上預約預約掛號系統的設計與實現 需求使用SSH框架(spring+struts2+hibernate)實現一個網上預約預約掛號系統, 用戶登錄註冊登錄系統, 能按科室查看醫生, 並能夠進行預約掛號和線上留言, 後臺管理系統更能夠進行科室管理，醫生管理，預約管理 ...
  • Java連載101-DataOutputStream、PrintStream方法詳解
    一、java.io.DataOutputStream；數據位元組輸出流 1.可以將記憶體中的“int i = 2;"寫入到硬碟文件裡面，寫進去的不是字元串，寫進去的是二進位數據，可以帶有類型。 package com.bjpowernode.java_learning; import java.io.* ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類