How we redesigned the NSQ- 其餘特性及將來計劃

在系列文章前面幾篇中，介紹了 NSQ 改造的過程和幾個基礎特性，本文中咱們繼續介紹幾個高級特性及其使用場景，這些都是結合有贊業務場景總結提煉出來的重要功能。

NSQ 拓展消息格式的設計

有贊中間件在 NSQ 中引入了支持拓展內容的消息格式，經過支持拓展的消息格式。業務方可以在消息體外定義額外的數據，拓展了應用功能，支持更多的場景。

相比較於 Kafka 等消息中間件，NSQ 的消息格式在內容和數量上較爲簡單。一條消息除了基本的元數據以外，其他內容爲消息體。消息的元數據主要包括了消息在服務端產生時的時間戳，服務端對於該消息的下發次數，消息 ID。Kafka消息格式(record batch,control record,record)中出現的部
分元數據例如壓縮格式（snappy），NSQ 在客戶端建連的過程當中經過 IDENTIFY 確認，而部分元數據，如 CRC，事務屬性等，在 NSQ 中則沒有對應實現。

消息格式的相對簡單，使得 NSQ 傳輸消息內容上有更高的效率，同時使得編寫 NSQ 客戶端時更爲容易。而簡單格式所帶來的缺點就是 NSQ 消息除了消息體自己以外，沒法攜帶更多的額外信息。在傳輸一些能夠和業務流程解耦的數據時，依然須要修改已有消息格式，而且因爲缺乏重用性，每一個須要傳輸拓展數據的業務方都須要從新改造本身的業務消息格式。

拓展內容的消息格式

爲了使 NSQ 支持更多的場景，有贊中間件在原有 NSQ 消息格式的基礎上進行了改進，設計並實現了一種支持拓展的消息格式。

能夠看到新消息格式在已有消息格式上增長了 3 個部分(綠色字體)：

• 拓展內容的版本（version of extension content）：

長度爲 1 個字節，用於區分拓展內容的類別和格式。例如，0x01 爲 json 拓展；

• 拓展內容的長度（length of extension content）：

長度爲 2 個字節，表示拓展內容的字節長度；

• 拓展內容的二進制字符串：可變長度，爲拓展內容的二進制字節數組；

經過在消息格式中引入以上附加信息，NSQ 在消息傳輸過程當中可以在不修改原有消息格式的前提下附帶額外的信息，業務方或者應用框架可以經過拓展消息格式支持新的場景和新的功能。在此咱們以有贊業務中使用的幾個典型場景爲例, 詳細描述下擴展消息的使用。

拓展消息使用場景之鏈路壓測

鏈路壓測是生產環境中的典型場景。壓測器在短期內生產大量線上壓測數據，用以檢測線上鏈路的性能以及可用性。針對壓測鏈路上使用消息中間件的應用，經過拓展消息設計，在鏈路壓測場景中，消息中間件能夠提供以下功能。

fig1. 消息使用場景之鏈路壓測

生產者應用在處理壓測消息時，在拓展消息頭中標記該消息爲壓測消息。NSQ 將線上消息以及壓測消息統一下發至下游消費者(線上 Consumer)，下游消費者經過檢查拓展消息中的壓測字段來判斷該消息是否爲壓測流量，由應用框架根據拓展消息頭內容決定是否下發至應用，或者對壓測消息進行攔截。

該方案的優點在於，應用方無需對已有 NSQ 的 topic 生產/消費配置進行變動，新版 NSQ 經過對已有 topic 進行升級，使 topic 支持拓展消息格式。業務方僅須要關注壓測消息的處理。該方案的缺點在於，線上消息和壓測消息共用一個 topic，未進行隔離。一旦生產者對於壓測消息的處理出現錯誤，或者下游消費應用超過負載時，此時隔離壓測數據的操做較爲複雜，須要業務方修改代碼，新版 NSQ 經過回溯消費功能來「洗掉」壓測消息。

拓展消息使用場景之鏈路隔離

拓展消息的另一種場景爲應用鏈路隔離。場景以下：QA 環境總存在兩類應用，第一類是 QA 環境中應用的穩定版本，另一類是應用在 QA 上進行新功能開發/驗證的版本。QA 環境中應用經過 NSQ 進行解耦。新功能版本中增長了新的消息處理邏輯來消費穩定 QA 環境中不支持的消息，在 NSQ 不支持鏈路隔離前，開發須要：

1. 中止 QA 穩定消費，啓動新功能驗證的消費；
2. 在 NSQ 上驗證新功能；
3. 中止新功能驗證消費，恢復穩定 QA 消費；
4. 以上步驟往復，直至原有 QA 被替換；

fig2. QA 環境中應用使用 NSQ 場景

經過在 NSQ 服務端實現基於拓展消息頭內容的投遞優先級，新版 NSQ 支持業務上鍊路隔離的需求。

fig3. 新版 NSQ 支持鏈路隔離應用場景

供新功能驗證的消息將經過在拓展消息頭上的附帶信息進行標記，NSQ 服務端在投遞消息時根據消息頭中的投遞信息(Tag)按照如下規則進行路由：

1. 消費者中不存在帶有相同投遞信息的消費者時，消息統一投遞給 QA 穩定環境的消費者;
2. 消費者中存在和消息頭中相同的投遞信息時，消息投遞給該消費者;
3. 消息投中不包含投遞信息時，消息統一投遞給 QA 穩定環境的消費者；

經過實現該規則，新版 NSQ 支持業務方實現環境鏈路隔離。

拓展消息使用場景之消息過濾

NSQ 消息的消費模式爲，消息在 channel 之間爲組播，channel 內的客戶端(Consumer)競爭一條消息。

fig4.NSQ 消息投遞機制

與鏈路隔離的思路相似，經過對消息拓展頭的指定值進行過濾，新版 NSQ 能夠支持 channel 內的消息過濾。

訂閱到相同 channel 上的消費者附帶相同的拓展消息關鍵字，當 NSQ 投遞消息時：

1. 消息內容沒有標識信息或者標識信息空, 則只會投遞沒有 filter_key 或者 filter_key 爲空的 channel;
2. 消息有過濾標識信息, 投遞到匹配的 filter_key 的消費 channel, 未指定 filter 的 channel 也要投遞;
3. 對於某個 channel 不匹配的消息, 服務端視爲已消費,現象爲該 channel 不投遞;

fig5. NSQ 基於 channel 的消息過濾

該功能的實現基於消息拓展頭，能夠在服務端，客戶端單獨實現，或由服務端和客戶端共同實現。

NSQ migrate proxy-nsq 遷移工具

對於正在使用開源版本 NSQ 的用戶，NSQ migrate proxy 提供將開源版本 NSQ 遷移到有贊自研版本 NSQ 的能力。藉助於該遷移工具，可在用戶無感知的狀況下對 topic 進行遷移。NSQ migrate proxy 在遷移過程當中做爲開源 NSQ 和自研 NSQ 的代理，根據遷移階段的變化將 lookup 請求代理至開源 NSQ 和自研 NSQ，整合 nsqlookupd 的結果後返回給客戶端。使用遷移代理須要鏈接客戶端實現讀寫策略，遷移代理須要根據讀（r）寫(w)參數對對生產者和消費者進行區分。

fig6. nsq遷移結構圖

遷移步驟

結合自研版 NSQ 的讀寫策略（r/w），NSQ migrate proxy定義了 3 個遷移階段，到達最後階段後，topic 的生產消費便遷移到自研版本

1.第 1 階段中，代理將在返回給客戶端的 lookup 結果中包含兩個 NSQ 集羣的節點信息。消費者將在兩個集羣間創建消費鏈接。生產繼續向開源 NSQ 進行生產。

fig7.遷移階段1

2.第 2 階段中，代理對於生產者的 lookup 請求，只返回遷移目標集羣的 lookup 結果。此時消息生產將指向目標 NSQ 集羣。消費者繼續維持雙集羣消費。

fig8.遷移階段2

3.當確認開源 NSQ 集羣中的消息已經消費完後，遷移進入最後階段。代理對於消費者的 lookup 請求只返回目標 NSQ 節點信息。消費和開源 NSQ 的鏈接將斷開。此時消息的生產和消費都遷移到自研 NSQ 集羣。遷移完成。

fig9.遷移階段3

Connectors

除了圍繞 NSQ 自己的的改造，咱們針對 spark 和 flume 嘗試了經過拓展與 NSQ 進行集成。

spark connectors

spark consumer 做爲 NSQ 的消費者，從 NSQ 消費消息後經過 spark streaming API 進行處理。

flume connectors

flume nsq sink 做爲 apache flume sink 拓展，用於鏈接 flume 和 NSQ，並經過本地文件序列化保存發送失敗的 event body 並重試。經過插件的方式，用戶在 flume 中的配置文件中指定 NSQ 做爲 flume 的下游。

將來計劃

爲了支撐更多樣的業務需求，有贊 NSQ 還在繼續完善和豐富更多新特性， 這些特性包括 NSQ 自己的特性開發，也包括基於 NSQ 作的外部擴展系統的開發。將來的一段時間，咱們計劃增長以下值得期待的重要特性。

流控

目前有贊有大量的 topic 都部署在一個大的集羣，受益於 golang 的goroutine模型，每一個topic基本都是獨立的處理，互相直接影響不大, 可是碰到一些數據量大的狀況, 仍是會對其餘topic形成必定的影響，特別是一些網絡流量很是大的 topic，爲了下降這種topic流量影響，咱們須要限制一些topic的流量上限， 對整個集羣的穩定性提供保障。 設計方案上， 咱們計劃使用業界經常使用的令牌桶方案。

批量訂閱

目前的 NSQ 仍是沿用每條消息 ack 的模式， 保持兼容特性。 性能上雖然知足目前以及將來一段時間的業務需求，可是還有改進的空間。特別是在某些網絡延遲較高的場景下，批量訂閱能夠大大提升吞吐量。批量訂閱將會支持一次消費一組消息而且能夠一次性 ack 一組消息，從而減小必定的網絡開銷。

更豐富安全審計功能

原版的 NSQ 已經支持一部分的安全審計功能, 包括使用安全連接以及使用驗證服務器，咱們後面將會針對 topic 的生產和 channel 的一些操做提供獨立的安全驗證服務，並作好審計日誌，防範一些安全問題。另外針對 nsqadmin 也會打通內部的統一登陸驗證，針對性的限制業務的一些危險操做。

分佈式事務協調器

微服務拆分的痛點就是多個系統之間的一致性保證問題，所以急需一個統一的框架能解決此類問題。分佈式事務協調器將會是構建在 NSQ 基礎之上的一個重要產品, 該產品將會充分利用 NSQ 的一些特性去解決業務的痛點。

基於消息內容的消費過濾

雖然目前已經有支持基於消息擴展頭進行初步過濾的功能，可是也有些業務需求很是定製化，須要更加複雜的過濾規則，這種狀況爲了不給 NSQ 核心代碼帶來影響，咱們也計劃在 NSQ 之上構建一個更加複雜的過濾系統去作和業務耦合的事情，避免給 NSQ 注入過多的業務耦合功能.

總結

本文中，先展現了有贊中間件在 NSQ 中引入的支持拓展的消息格式，並經過 3 個業務場景來展現新的消息格式的玩法。以後的部分介紹了圍繞自研版本 NSQ 開發的拓展工具，包括了用於遷移的代理，以及能夠將 NSQ 與 spark 和 flume 進行集成的拓展。最後對於將來計劃進行了介紹，展望了部分計劃中的新特性。

參考資料

• [1]NSQ spark consumer: https://github.com/youzan/spa...
• [2]NSQ flume sink: https://github.com/DoraALin/f...

點擊連接，可直達《How we redesigned the NSQ》系列全部文章：
https://tech.youzan.com/tag/p...