精華！20k Start的消息隊列NSQ源碼剖析

什麼是NSQ

NSQ is a realtime distributed messaging platform designed to operate at scale, handling billions of messages per day    

與kafka、MQ爲同一類消息中間件，基於Go語言實現而且因爲輕量化，備受好評
它基於MIT開源協議發佈，由bitly公司開源出來的一款簡單易用的消息中間件
複製代碼

術語介紹

Topic ：一個topic就是程序發佈消息的一個邏輯鍵，當程序第一次發佈消息時就會建立topic
Channels ：channel與消費者相關，是消費者之間的負載均衡，channel在某種意義上來講是一個「隊列」

nsqd：nsqd 是一個守護進程，負責接收，排隊，投遞消息給客戶端
nsqlookupd：nsqlookupd 是守護進程負責管理拓撲信息
nsqadmin：nsqadmin 是一套 WEB UI，用來聚集集羣的實時統計，並執行不一樣的管理任務
nsq_to _file：消費指定的話題（topic）/通道（channel），並寫到文件中，有選擇的滾動和/或壓縮文件
nsq_to _http：消費指定的話題（topic）/通道（channel）和執行 HTTP requests (GET/POST) 到指定的端點
nsq_to _nsq：消費者指定的話題/通道和重發布消息到目的地 nsqd 經過 TCP
複製代碼

NSQ的應用場景

流程異步化、代碼解耦合、流量削峯、高可用、高吞吐量、廣播分發，達到數據的最終一致性，知足具體的業務場景需求
複製代碼

從該項目中你能得到什麼？

Golang的入門    
複製代碼

主題

• NSQ系統架構

  會由三個組件構成，分別扮演不一樣的角色
  重要性： NSQ > NSQLOOKUP > NSD Admin

  1. NSQ
  2. NSQLOOKUP
  3. NSD Admin

  爲了便於你們理解，能夠當作坐飛機。
  把NSQ看做飛機，NSQLOOKUP當作飛機的調度臺，NSQ Admin爲訂票的app，而咱們本身則是整個過程中的message。NSQ會搭載着咱們去想去的地方，航班號就是topic。

• NSQ的代碼結構

  這裏也大體分爲3塊
  1.入口函數，build app
  2.公共的抽象
  3.nsq、nsqlookup、admin的service層，實現實際的func，後續將對於這部分進行重點展開

• NSQ的組件剖析

• nsqd

  nsqd 是一個守護進程，負責接收（生產者 producer ）、排隊（最小堆實現）、投遞（消費者 consumer ）消息給客戶端。它能夠獨立運行，不過一般它是由 nsqlookupd 實例所在集羣配置的

  nsqd的執行邏輯

  Producer --> 消息 --> topic --> channels --> Consumer

  nsqd對象

  TCP 服務    
    HTTP 服務
  
    圖中能夠看到，核心對象有2個，topicMap 和 Listener    
    topicMap存儲topic信息，tcpListener 和httpListener監聽tcp服務和http服務    
    其餘的對象則是圍繞此展開，例如：有保證上下文的context，有原子性的sync.RWMutex，有消息池管理的poolSize等
  複製代碼

  nsqd 的啓動流程

  1，會啓動tcp server 和http server ，並以daemon的方式進行監聽 2，會不斷給nsqlookupd發送註冊心跳，保證在nsqdlookup註冊成功且存活

  tips:這裏nsqd藉助了第三方的svc包，對後臺進程進行了管理
  "github.com/judwhite/go-svc" 這個是一個服務運行框架

  消息的傳遞

  nsqd是如何進行消息的分發傳遞的？ 分發的過程當中應該注意什麼問題？

  咱們先來看看nsqd對於消息的處理過程
  生產者，先將消息傳遞給Topics，達到A，而後會將指定的消息分發給特定的Channel，而後就會經過這個channel進行消息的消費，正因如此，也實現了消費者的負載均衡

  1, 消息的內存優化
  經過sync.Pool 保存複用 臨時對象，避免存儲從新分配，下降程序GC的壓力，從而避免：併發⼤－佔⽤內存⼤－GC 緩慢－處理併發能⼒下降－併發更⼤

  2，消息的傳遞過程對象
  使用byte來進行消息的傳遞，避免頻繁格式轉化致使的性能下降

  3，消息的格式
  時間戳（8byte）+msgID（16byte）+body（*byte）

  4，消息的高可靠性
  超出閾值的會有內存轉爲磁盤進行持久化

  5,經過最小堆(min heap)進行channel的排隊

  nsqd 的接收消息過程

  1，由一個協程來監聽tcp server
    2，由一個協程來處理請求來的client，並保持長連接，當conn failed的時候，還會主動讓出cpu
    3，由一個協程來監聽tcp socket，並持續對消息進行處理
  複製代碼

  [tcp server 的監聽]

  [持續處理來自client的請求]

• NSQLOOKUP

  nsqlookupd 是守護進程負責管理拓撲信息。客戶端經過查詢 nsqlookupd 來發現指定話題（ topic ）的生產者，而且 nsqd 節點廣播話題（topic）和通道（ channel ）信息。有兩個接口： TCP 接口， nsqd 用它來廣播。 HTTP 接口，客戶端用它來發現和管理

  NSQLOOKUP對象則較爲簡單
  1個原子性和6個對象，其中分別監聽tcp和http 服務

  • TCP服務：
  • HTTP服務：

• NSQAdmin

  nsqadmin 是一套 WEB UI，用來聚集集羣的實時統計，並執行不一樣的管理任務
    因爲這塊內容比較單一，不進行過多的解析
  複製代碼

亮點

性能優化手段

- sync.pool 進行內存複用,內存分片持久化，避免頻繁回收致使的GC太高    
   - byte流傳遞，避免頻繁序列化致使的性能降低    
   - cpu讓出，例如，當conn err時候，主動讓出cpu
   - 最小堆排序
   - 並發動態調整
   - 機率過時算法
   - 自定義數據報文
    
複製代碼

開發手段

- 設計模式：工廠方法、裝飾器等    
   - Go 併發處理，動態調整goroutiune
   - 優雅的OOP思路
複製代碼

總結

雖然NSQ並不是完美，總體架構也並不是複雜，本篇介紹的內容也並不全面
可是，細節之處請細品

附錄

NSQ的 Git：github.com/nsqio/nsq ，做爲一個擁有20k star的消息中間件