1. 介紹
目前成都創(chuàng)新互聯(lián)公司已為數(shù)千家的企業(yè)提供了網(wǎng)站建設(shè)、域名、網(wǎng)站空間、網(wǎng)站托管運營、企業(yè)網(wǎng)站設(shè)計、遂川網(wǎng)站維護等服務(wù)，公司將堅持客戶導向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。
最近在研究一些消息中間件，常用的MQ如RabbitMQ,ActiveMQ,Kafka等。NSQ是一個基于Go語言的分布式實時消息平臺，它基于MIT開源協(xié)議發(fā)布，由bitly公司開源出來的一款簡單易用的消息中間件。
官方和第三方還為NSQ開發(fā)了眾多客戶端功能庫，如官方提供的基于HTTP的nsqd、Go客戶端go-nsq、Python客戶端pynsq、基于Node.js的JavaScript客戶端nsqjs、異步C客戶端libnsq、Java客戶端nsq-java以及基于各種語言的眾多第三方客戶端功能庫。
1.1 Features
1). Distributed
NSQ提供了分布式的，去中心化，且沒有單點故障的拓撲結(jié)構(gòu)，穩(wěn)定的消息傳輸發(fā)布保障，能夠具有高容錯和HA（高可用）特性。
2). Scalable易于擴展
NSQ支持水平擴展，沒有中心化的brokers。內(nèi)置的發(fā)現(xiàn)服務(wù)簡化了在集群中增加節(jié)點。同時支持pub-sub和load-balanced 的消息分發(fā)。
3). Ops Friendly
NSQ非常容易配置和部署，生來就綁定了一個管理界面。二進制包沒有運行時依賴。官方有Docker image。
4.Integrated高度集成
官方的 Go 和 Python庫都有提供。而且為大多數(shù)語言提供了庫。
1.2 組件
1.3 拓撲結(jié)構(gòu)
NSQ推薦通過他們相應(yīng)的nsqd實例使用協(xié)同定位發(fā)布者，這意味著即使面對網(wǎng)絡(luò)分區(qū)，消息也會被保存在本地，直到它們被一個消費者讀取。更重要的是，發(fā)布者不必去發(fā)現(xiàn)其他的nsqd節(jié)點，他們總是可以向本地實例發(fā)布消息。
NSQ
首先，一個發(fā)布者向它的本地nsqd發(fā)送消息，要做到這點，首先要先打開一個連接，然后發(fā)送一個包含topic和消息主體的發(fā)布命令，在這種情況下，我們將消息發(fā)布到事件topic上以分散到我們不同的worker中。
事件topic會復(fù)制這些消息并且在每一個連接topic的channel上進行排隊，在我們的案例中，有三個channel，它們其中之一作為檔案channel。消費者會獲取這些消息并且上傳到S3。
nsqd
每個channel的消息都會進行排隊，直到一個worker把他們消費，如果此隊列超出了內(nèi)存限制，消息將會被寫入到磁盤中。Nsqd節(jié)點首先會向nsqlookup廣播他們的位置信息，一旦它們注冊成功，worker將會從nsqlookup服務(wù)器節(jié)點上發(fā)現(xiàn)所有包含事件topic的nsqd節(jié)點。
nsqlookupd
2. Internals
2.1 消息傳遞擔保
1）客戶表示已經(jīng)準備好接收消息
2）NSQ 發(fā)送一條消息，并暫時將數(shù)據(jù)存儲在本地（在 re-queue 或 timeout）
3）客戶端回復(fù) FIN（結(jié)束）或 REQ（重新排隊）分別指示成功或失敗。如果客戶端沒有回復(fù), NSQ 會在設(shè)定的時間超時，自動重新排隊消息
這確保了消息丟失唯一可能的情況是不正常結(jié)束 nsqd 進程。在這種情況下，這是在內(nèi)存中的任何信息（或任何緩沖未刷新到磁盤）都將丟失。
如何防止消息丟失是最重要的，即使是這個意外情況可以得到緩解。一種解決方案是構(gòu)成冗余 nsqd對（在不同的主機上）接收消息的相同部分的副本。因為你實現(xiàn)的消費者是冪等的，以兩倍時間處理這些消息不會對下游造成影響，并使得系統(tǒng)能夠承受任何單一節(jié)點故障而不會丟失信息。
2.2 簡化配置和管理
單個 nsqd 實例被設(shè)計成可以同時處理多個數(shù)據(jù)流。流被稱為“話題”和話題有 1 個或多個“通道”。每個通道都接收到一個話題中所有消息的拷貝。在實踐中，一個通道映射到下行服務(wù)消費一個話題。
在更底的層面，每個 nsqd 有一個與 nsqlookupd 的長期 TCP 連接，定期推動其狀態(tài)。這個數(shù)據(jù)被 nsqlookupd 用于給消費者通知 nsqd 地址。對于消費者來說，一個暴露的 HTTP /lookup 接口用于輪詢。為話題引入一個新的消費者，只需啟動一個配置了 nsqlookup 實例地址的 NSQ 客戶端。無需為添加任何新的消費者或生產(chǎn)者更改配置，大大降低了開銷和復(fù)雜性。
2.3 消除單點故障
NSQ被設(shè)計以分布的方式被使用。nsqd 客戶端（通過 TCP ）連接到指定話題的所有生產(chǎn)者實例。沒有中間人，沒有消息代理，也沒有單點故障。
這種拓撲結(jié)構(gòu)消除單鏈，聚合，反饋。相反，你的消費者直接訪問所有生產(chǎn)者。從技術(shù)上講，哪個客戶端連接到哪個 NSQ 不重要，只要有足夠的消費者連接到所有生產(chǎn)者，以滿足大量的消息，保證所有東西最終將被處理。對于 nsqlookupd，高可用性是通過運行多個實例來實現(xiàn)。他們不直接相互通信和數(shù)據(jù)被認為是最終一致。消費者輪詢所有的配置的 nsqlookupd 實例和合并 response。失敗的，無法訪問的，或以其他方式故障的節(jié)點不會讓系統(tǒng)陷于停頓。
2.4 效率
對于數(shù)據(jù)的協(xié)議，通過推送數(shù)據(jù)到客戶端最大限度地提高性能和吞吐量的，而不是等待客戶端拉數(shù)據(jù)。這個概念，稱之為 RDY 狀態(tài)，基本上是客戶端流量控制的一種形式。
efficiency
2.5 心跳和超時
組合應(yīng)用級別的心跳和 RDY 狀態(tài)，避免頭阻塞現(xiàn)象，也可能使心跳無用（即，如果消費者是在后面的處理消息流的接收緩沖區(qū)中，操作系統(tǒng)將被填滿，堵心跳）為了保證進度，所有的網(wǎng)絡(luò) IO 時間上限勢必與配置的心跳間隔相關(guān)聯(lián)。這意味著，你可以從字面上拔掉之間的網(wǎng)絡(luò)連接 nsqd 和消費者，它會檢測并正確處理錯誤。當檢測到一個致命錯誤，客戶端連接被強制關(guān)閉。在傳輸中的消息會超時而重新排隊等待傳遞到另一個消費者。最后，錯誤會被記錄并累計到各種內(nèi)部指標。
2.6 分布式
因為NSQ沒有在守護程序之間共享信息，所以它從一開始就是為了分布式操作而生。個別的機器可以隨便宕機隨便啟動而不會影響到系統(tǒng)的其余部分，消息發(fā)布者可以在本地發(fā)布，即使面對網(wǎng)絡(luò)分區(qū)。
這種“分布式優(yōu)先”的設(shè)計理念意味著NSQ基本上可以永遠不斷地擴展，需要更高的吞吐量？那就添加更多的nsqd吧。唯一的共享狀態(tài)就是保存在lookup節(jié)點上，甚至它們不需要全局視圖，配置某些nsqd注冊到某些lookup節(jié)點上這是很簡單的配置，唯一關(guān)鍵的地方就是消費者可以通過lookup節(jié)點獲取所有完整的節(jié)點集。清晰的故障事件——NSQ在組件內(nèi)建立了一套明確關(guān)于可能導致故障的的故障權(quán)衡機制，這對消息傳遞和恢復(fù)都有意義。雖然它們可能不像Kafka系統(tǒng)那樣提供嚴格的保證級別，但NSQ簡單的操作使故障情況非常明顯。
2.7 no replication
不像其他的隊列組件，NSQ并沒有提供任何形式的復(fù)制和集群，也正是這點讓它能夠如此簡單地運行，但它確實對于一些高保證性高可靠性的消息發(fā)布沒有足夠的保證。我們可以通過降低文件同步的時間來部分避免，只需通過一個標志配置，通過EBS支持我們的隊列。但是這樣仍然存在一個消息被發(fā)布后馬上死亡，丟失了有效的寫入的情況。
2.8 沒有嚴格的順序
雖然Kafka由一個有序的日志構(gòu)成，但NSQ不是。消息可以在任何時間以任何順序進入隊列。在我們使用的案例中，這通常沒有關(guān)系，因為所有的數(shù)據(jù)都被加上了時間戳，但它并不適合需要嚴格順序的情況。
2.9 無數(shù)據(jù)重復(fù)刪除功能
NSQ對于超時系統(tǒng)，它使用了心跳檢測機制去測試消費者是否存活還是死亡。很多原因會導致我們的consumer無法完成心跳檢測，所以在consumer中必須有一個單獨的步驟確保冪等性。
3. 實踐安裝過程
本文將nsq集群具體的安裝過程略去，大家可以自行參考官網(wǎng)，比較簡單。這部分介紹下筆者實驗的拓撲，以及nsqadmin的相關(guān)信息。
3.1 拓撲結(jié)構(gòu)
topology
實驗采用3臺NSQD服務(wù)，2臺LOOKUPD服務(wù)。
采用官方推薦的拓撲，消息發(fā)布的服務(wù)和NSQD在一臺主機。一共5臺機器。
NSQ基本沒有配置文件，配置通過命令行指定參數(shù)。
主要命令如下:
LOOKUPD命令
NSQD命令
工具類，消費后存儲到本地文件。
發(fā)布一條消息
3.2 nsqadmin
對Streams的詳細信息進行查看，包括NSQD節(jié)點，具體的channel，隊列中的消息數(shù)，連接數(shù)等信息。
nsqadmin
channel
列出所有的NSQD節(jié)點:
nodes
消息的統(tǒng)計:
msgs
lookup主機的列表:
hosts
4. 總結(jié)
NSQ基本核心就是簡單性，是一個簡單的隊列，這意味著它很容易進行故障推理和很容易發(fā)現(xiàn)bug。消費者可以自行處理故障事件而不會影響系統(tǒng)剩下的其余部分。
事實上，簡單性是我們決定使用NSQ的首要因素，這方便與我們的許多其他軟件一起維護，通過引入隊列使我們得到了堪稱完美的表現(xiàn)，通過隊列甚至讓我們增加了幾個數(shù)量級的吞吐量。越來越多的consumer需要一套嚴格可靠性和順序性保障，這已經(jīng)超過了NSQ提供的簡單功能。
結(jié)合我們的業(yè)務(wù)系統(tǒng)來看，對于我們所需要傳輸?shù)陌l(fā)票消息，相對比較敏感，無法容忍某個nsqd宕機，或者磁盤無法使用的情況，該節(jié)點堆積的消息無法找回。這是我們沒有選擇該消息中間件的主要原因。簡單性和可靠性似乎并不能完全滿足。相比Kafka，ops肩負起更多負責的運營。另一方面，它擁有一個可復(fù)制的、有序的日志可以提供給我們更好的服務(wù)。但對于其他適合NSQ的consumer，它為我們服務(wù)的相當好，我們期待著繼續(xù)鞏固它的堅實的基礎(chǔ)。
很多朋友可能知道Go語言的優(yōu)勢在哪，卻不知道Go語言適合用于哪些地方。
1、 Go語言作為服務(wù)器編程語言，很適合處理日志、數(shù)據(jù)打包、虛擬機處理、文件系統(tǒng)、分布式系統(tǒng)、數(shù)據(jù)庫代理等；網(wǎng)絡(luò)編程方面。Go語言廣泛應(yīng)用于Web應(yīng)用、API應(yīng)用、下載應(yīng)用等；除此之外，Go語言還可用于內(nèi)存數(shù)據(jù)庫和云平臺領(lǐng)域，目前國外很多云平臺都是采用Go開發(fā)。
2、 其實Go語言主要用作服務(wù)器端開發(fā)。其定位是用來開發(fā)"大型軟件"的，適合于很多程序員一起開發(fā)大型軟件，并且開發(fā)周期長，支持云計算的網(wǎng)絡(luò)服務(wù)。Go語言能夠讓程序員快速開發(fā)，并且在軟件不斷的增長過程中，它能讓程序員更容易地進行維護和修改。它融合了傳統(tǒng)編譯型語言的高效性和腳本語言的易用性和富于表達性。
3、 Go語言成功案例。Nsq：Nsq是由Go語言開發(fā)的高性能、高可用消息隊列系統(tǒng)，性能非常高，每天能處理數(shù)十億條的消息；
4、 Docker：基于lxc的一個虛擬打包工具，能夠?qū)崿F(xiàn)PAAS平臺的組建。
5、 Packer：用來生成不同平臺的鏡像文件，例如VM、vbox、AWS等，作者是vagrant的作者
6、 Skynet：分布式調(diào)度框架。
7、 Doozer：分布式同步工具，類似ZooKeeper。
8、 Heka：mazila開源的日志處理系統(tǒng)。
9、 Cbfs：couchbase開源的分布式文件系統(tǒng)。
10、 Tsuru：開源的PAAS平臺，和SAE實現(xiàn)的功能一模一樣。
11、 Groupcache：memcahe作者寫的用于Google下載系統(tǒng)的緩存系統(tǒng)。
12、 God：類似redis的緩存系統(tǒng)，但是支持分布式和擴展性。
13、 Gor：網(wǎng)絡(luò)流量抓包和重放工具。
以上的就是關(guān)于go語言能做什么的內(nèi)容介紹了。
目前websocket技術(shù)已經(jīng)很成熟，選型Go語言，當然是為了節(jié)省成本以及它強大的高并發(fā)性能。我使用的是第三方開源的websocket庫即gorilla/websocket。
由于我們線上推送的量不小，推送后端需要部署多節(jié)點保持高可用，所以需要自己做集群，具體架構(gòu)方案如圖：
Auth Service：鑒權(quán)服務(wù)，根據(jù)Token驗證用戶權(quán)限。
Collect Service：消息采集服務(wù)，負責收集業(yè)務(wù)系統(tǒng)消息，存入MongoDB后，發(fā)送給消息分發(fā)服務(wù)。
Dispatch Service：消息分發(fā)服務(wù)，根據(jù)路由規(guī)則分發(fā)至對應(yīng)消息推送服務(wù)節(jié)點上。
Push Service：消息推送服務(wù)，通過websocket將消息推送給用戶。
集群推送的關(guān)鍵點在于，web端與服務(wù)端建立長連接之后，具體跟哪個推送節(jié)點保持長連接的，如果我們能夠找到對應(yīng)的連接節(jié)點，那么我們就可以將消息推送出去。下面講解一下集群的大致流程：
1. web端用戶登錄之后，帶上token與后端推送服務(wù)（Push Service）保持長連接。
2. 推送服務(wù)收到連接請求之后，攜帶token去鑒權(quán)服務(wù)（Auth Service）驗證此token權(quán)限，并返回用戶ID。
3. 把返回的用戶ID與長連接存入本地緩存，保持用戶ID與長連接綁定關(guān)系。
4. 再將用戶ID與本推送節(jié)點IP存入redis，建立用戶（即長連接）與節(jié)點綁定關(guān)系，并設(shè)置失效時間。
5. 采集服務(wù)（Collect Service）收集業(yè)務(wù)消息，首先存入mongodb，然后將消息透傳給分發(fā)服務(wù)（Dispatch Service）。
6. 分發(fā)服務(wù)收到消息之后，根據(jù)消息體中的用戶ID，從redis中獲取對應(yīng)的推送服務(wù)節(jié)點IP，然后轉(zhuǎn)發(fā)給對應(yīng)的推送節(jié)點。
7. 推送服務(wù)節(jié)點收到消息之后，根據(jù)用戶ID，從本地緩存中取出對應(yīng)的長連接，將消息推送給客戶端。
其他注意事項：
上一節(jié)中，我們?yōu)槊總€連接都創(chuàng)建了一個goroutine來讀取其中的消息，現(xiàn)在我們將這個讀取消息的方法實現(xiàn)一下。
我們在application目錄下新建controllers目錄，并在其中創(chuàng)建一個MessageController.go文件。
首先我們新建一個MessageController的結(jié)構(gòu)體，內(nèi)容如下
這個結(jié)構(gòu)體包括兩個內(nèi)容，一個是我們將連接放在數(shù)組之后，返回的索引，另一個是連接本身.
這個是具體的方法。
我們首先設(shè)置了一下讀消息的大小、超時時間以及超時后需要的操作。
超時時間如果設(shè)置為0,那么就是永不超時。之前在這里直接寫0，被告知需要傳一個time.Time類型的數(shù)據(jù)。最終谷歌后才得到了這個值time.Time{}為"0001-01-01 00:00:00 +0000 UTC"。
我們將用戶手法消息的內(nèi)容定義為一個結(jié)構(gòu)體，然后將用戶的訂閱信息的json通過json.unmarshal轉(zhuǎn)換成這個結(jié)構(gòu)體。
之后的switch操作與我們在Swoole中的操作基本雷同，在查詢到login之后，調(diào)用service中 的login方法來進行注冊。
下一節(jié)中我們再介紹具體的注冊邏輯。
重慶分公司
重慶分公司
