摘要：一條消息除了基本的元數據之外，其余內容為消息體。消息的元數據主要包括了消息在服務端產生時的時間戳，服務端對于該消息的下發次數，消息。作為的消費者，從消費消息后通過進行處理。

在系列文章前面幾篇中，介紹了 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 不支持鏈路隔離前，開發需要：

停止 QA 穩定消費，啟動新功能驗證的消費；

在 NSQ 上驗證新功能；

停止新功能驗證消費，恢復穩定 QA 消費；

以上步驟往復，直至原有 QA 被替換；

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

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

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

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

消費者中不存在帶有相同投遞信息的消費者時，消息統一投遞給 QA 穩定環境的消費者;

消費者中存在和消息頭中相同的投遞信息時，消息投遞給該消費者;

消息投中不包含投遞信息時，消息統一投遞給 QA 穩定環境的消費者；

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

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

fig4.NSQ 消息投遞機制

與鏈路隔離的思路類似，通過對消息拓展頭的指定值進行過濾，新版 NSQ 可以支持 channel 內的消息過濾。

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

消息內容沒有標識信息或者標識信息空, 則只會投遞沒有 filter_key 或者 filter_key 為空的 channel;

消息有過濾標識信息, 投遞到匹配的 filter_key 的消費 channel, 未指定 filter 的 channel 也要投遞;

對于某個 channel 不匹配的消息, 服務端視為已消費,現象為該 channel 不投遞;

fig5. NSQ 基于 channel 的消息過濾

該功能的實現基于消息拓展頭，可以在服務端，客戶端多帶帶實現，或由服務端和客戶端共同實現。

對于正在使用開源版本 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

除了圍繞 NSQ 本身的的改造，我們針對 spark 和 flume 嘗試了通過拓展與 NSQ 進行集成。

spark consumer 作為 NSQ 的消費者，從 NSQ 消費消息后通過 spark streaming API 進行處理。

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...