摘要：一條消息除了基本的元數(shù)據(jù)之外，其余內(nèi)容為消息體。消息的元數(shù)據(jù)主要包括了消息在服務(wù)端產(chǎn)生時(shí)的時(shí)間戳，服務(wù)端對于該消息的下發(fā)次數(shù)，消息。作為的消費(fèi)者，從消費(fèi)消息后通過進(jìn)行處理。

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

有贊中間件在 NSQ 中引入了支持拓展內(nèi)容的消息格式，通過支持拓展的消息格式。業(yè)務(wù)方能夠在消息體外定義額外的數(shù)據(jù)，拓展了應(yīng)用功能，支持更多的場景。

相比較于 Kafka 等消息中間件，NSQ 的消息格式在內(nèi)容和數(shù)量上較為簡單。一條消息除了基本的元數(shù)據(jù)之外，其余內(nèi)容為消息體。消息的元數(shù)據(jù)主要包括了消息在服務(wù)端產(chǎn)生時(shí)的時(shí)間戳，服務(wù)端對于該消息的下發(fā)次數(shù)，消息 ID。Kafka消息格式(record batch,control record,record)中出現(xiàn)的部
分元數(shù)據(jù)例如壓縮格式（snappy），NSQ 在客戶端建連的過程中通過 IDENTIFY 確認(rèn)，而部分元數(shù)據(jù)，如 CRC，事務(wù)屬性等，在 NSQ 中則沒有對應(yīng)實(shí)現(xiàn)。

消息格式的相對簡單，使得 NSQ 傳輸消息內(nèi)容上有更高的效率，同時(shí)使得編寫 NSQ 客戶端時(shí)更為容易。而簡單格式所帶來的缺點(diǎn)就是 NSQ 消息除了消息體本身之外，無法攜帶更多的額外信息。在傳輸一些可以和業(yè)務(wù)流程解耦的數(shù)據(jù)時(shí)，依然需要修改已有消息格式，并且由于缺少重用性，每個(gè)需要傳輸拓展數(shù)據(jù)的業(yè)務(wù)方都需要重新改造自己的業(yè)務(wù)消息格式。

為了使 NSQ 支持更多的場景，有贊中間件在原有 NSQ 消息格式的基礎(chǔ)上進(jìn)行了改進(jìn)，設(shè)計(jì)并實(shí)現(xiàn)了一種支持拓展的消息格式。

可以看到新消息格式在已有消息格式上增加了 3 個(gè)部分(綠色字體)：

拓展內(nèi)容的版本（version of extension content）：

長度為 1 個(gè)字節(jié)，用于區(qū)分拓展內(nèi)容的類別和格式。例如，0x01 為 json 拓展；

拓展內(nèi)容的長度（length of extension content）：

長度為 2 個(gè)字節(jié)，表示拓展內(nèi)容的字節(jié)長度；

拓展內(nèi)容的二進(jìn)制字符串：可變長度，為拓展內(nèi)容的二進(jìn)制字節(jié)數(shù)組；

通過在消息格式中引入以上附加信息，NSQ 在消息傳輸過程中能夠在不修改原有消息格式的前提下附帶額外的信息，業(yè)務(wù)方或者應(yīng)用框架能夠通過拓展消息格式支持新的場景和新的功能。在此我們以有贊業(yè)務(wù)中使用的幾個(gè)典型場景為例, 詳細(xì)描述下擴(kuò)展消息的使用。

鏈路壓測是生產(chǎn)環(huán)境中的典型場景。壓測器在短時(shí)間內(nèi)生產(chǎn)大量線上壓測數(shù)據(jù)，用以檢測線上鏈路的性能以及可用性。針對壓測鏈路上使用消息中間件的應(yīng)用，通過拓展消息設(shè)計(jì)，在鏈路壓測場景中，消息中間件可以提供如下功能。

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

生產(chǎn)者應(yīng)用在處理壓測消息時(shí)，在拓展消息頭中標(biāo)記該消息為壓測消息。NSQ 將線上消息以及壓測消息統(tǒng)一下發(fā)至下游消費(fèi)者(線上 Consumer)，下游消費(fèi)者通過檢查拓展消息中的壓測字段來判斷該消息是否為壓測流量，由應(yīng)用框架根據(jù)拓展消息頭內(nèi)容決定是否下發(fā)至應(yīng)用，或者對壓測消息進(jìn)行攔截。

該方案的優(yōu)勢在于，應(yīng)用方無需對已有 NSQ 的 topic 生產(chǎn)/消費(fèi)配置進(jìn)行變更，新版 NSQ 通過對已有 topic 進(jìn)行升級，使 topic 支持拓展消息格式。業(yè)務(wù)方僅需要關(guān)注壓測消息的處理。該方案的缺點(diǎn)在于，線上消息和壓測消息共用一個(gè) topic，未進(jìn)行隔離。一旦生產(chǎn)者對于壓測消息的處理出現(xiàn)錯(cuò)誤，或者下游消費(fèi)應(yīng)用超過負(fù)載時(shí)，此時(shí)隔離壓測數(shù)據(jù)的操作較為復(fù)雜，需要業(yè)務(wù)方修改代碼，新版 NSQ 通過回溯消費(fèi)功能來“洗掉”壓測消息。

拓展消息的另外一種場景為應(yīng)用鏈路隔離。場景如下：QA 環(huán)境總存在兩類應(yīng)用，第一類是 QA 環(huán)境中應(yīng)用的穩(wěn)定版本，另外一類是應(yīng)用在 QA 上進(jìn)行新功能開發(fā)/驗(yàn)證的版本。QA 環(huán)境中應(yīng)用通過 NSQ 進(jìn)行解耦。新功能版本中增加了新的消息處理邏輯來消費(fèi)穩(wěn)定 QA 環(huán)境中不支持的消息，在 NSQ 不支持鏈路隔離前，開發(fā)需要：

停止 QA 穩(wěn)定消費(fèi)，啟動(dòng)新功能驗(yàn)證的消費(fèi)；

在 NSQ 上驗(yàn)證新功能；

停止新功能驗(yàn)證消費(fèi)，恢復(fù)穩(wěn)定 QA 消費(fèi)；

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

fig2. QA 環(huán)境中應(yīng)用使用 NSQ 場景

通過在 NSQ 服務(wù)端實(shí)現(xiàn)基于拓展消息頭內(nèi)容的投遞優(yōu)先級，新版 NSQ 支持業(yè)務(wù)上鏈路隔離的需求。

fig3. 新版 NSQ 支持鏈路隔離應(yīng)用場景

供新功能驗(yàn)證的消息將通過在拓展消息頭上的附帶信息進(jìn)行標(biāo)記，NSQ 服務(wù)端在投遞消息時(shí)根據(jù)消息頭中的投遞信息(Tag)按照以下規(guī)則進(jìn)行路由：

消費(fèi)者中不存在帶有相同投遞信息的消費(fèi)者時(shí)，消息統(tǒng)一投遞給 QA 穩(wěn)定環(huán)境的消費(fèi)者;

消費(fèi)者中存在和消息頭中相同的投遞信息時(shí)，消息投遞給該消費(fèi)者;

消息投中不包含投遞信息時(shí)，消息統(tǒng)一投遞給 QA 穩(wěn)定環(huán)境的消費(fèi)者；

通過實(shí)現(xiàn)該規(guī)則，新版 NSQ 支持業(yè)務(wù)方實(shí)現(xiàn)環(huán)境鏈路隔離。

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

fig4.NSQ 消息投遞機(jī)制

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

訂閱到相同 channel 上的消費(fèi)者附帶相同的拓展消息關(guān)鍵字，當(dāng) NSQ 投遞消息時(shí)：

消息內(nèi)容沒有標(biāo)識(shí)信息或者標(biāo)識(shí)信息空, 則只會(huì)投遞沒有 filter_key 或者 filter_key 為空的 channel;

消息有過濾標(biāo)識(shí)信息, 投遞到匹配的 filter_key 的消費(fèi) channel, 未指定 filter 的 channel 也要投遞;

對于某個(gè) channel 不匹配的消息, 服務(wù)端視為已消費(fèi),現(xiàn)象為該 channel 不投遞;

fig5. NSQ 基于 channel 的消息過濾

該功能的實(shí)現(xiàn)基于消息拓展頭，可以在服務(wù)端，客戶端多帶帶實(shí)現(xiàn)，或由服務(wù)端和客戶端共同實(shí)現(xiàn)。

對于正在使用開源版本 NSQ 的用戶，NSQ migrate proxy 提供將開源版本 NSQ 遷移到有贊自研版本 NSQ 的能力。借助于該遷移工具，可在用戶無感知的情況下對 topic 進(jìn)行遷移。NSQ migrate proxy 在遷移過程中作為開源 NSQ 和自研 NSQ 的代理，根據(jù)遷移階段的變化將 lookup 請求代理至開源 NSQ 和自研 NSQ，整合 nsqlookupd 的結(jié)果后返回給客戶端。使用遷移代理需要連接客戶端實(shí)現(xiàn)讀寫策略，遷移代理需要根據(jù)讀（r）寫(w)參數(shù)對對生產(chǎn)者和消費(fèi)者進(jìn)行區(qū)分。

fig6. nsq遷移結(jié)構(gòu)圖

結(jié)合自研版 NSQ 的讀寫策略（r/w），NSQ migrate proxy定義了 3 個(gè)遷移階段，到達(dá)最后階段后，topic 的生產(chǎn)消費(fèi)便遷移到自研版本

1.第 1 階段中，代理將在返回給客戶端的 lookup 結(jié)果中包含兩個(gè) NSQ 集群的節(jié)點(diǎn)信息。消費(fèi)者將在兩個(gè)集群間建立消費(fèi)連接。生產(chǎn)繼續(xù)向開源 NSQ 進(jìn)行生產(chǎn)。

fig7.遷移階段1

2.第 2 階段中，代理對于生產(chǎn)者的 lookup 請求，只返回遷移目標(biāo)集群的 lookup 結(jié)果。此時(shí)消息生產(chǎn)將指向目標(biāo) NSQ 集群。消費(fèi)者繼續(xù)維持雙集群消費(fèi)。

fig8.遷移階段2

3.當(dāng)確認(rèn)開源 NSQ 集群中的消息已經(jīng)消費(fèi)完后，遷移進(jìn)入最后階段。代理對于消費(fèi)者的 lookup 請求只返回目標(biāo) NSQ 節(jié)點(diǎn)信息。消費(fèi)和開源 NSQ 的連接將斷開。此時(shí)消息的生產(chǎn)和消費(fèi)都遷移到自研 NSQ 集群。遷移完成。

fig9.遷移階段3

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

spark consumer 作為 NSQ 的消費(fèi)者，從 NSQ 消費(fèi)消息后通過 spark streaming API 進(jìn)行處理。

flume nsq sink 作為 apache flume sink 拓展，用于連接 flume 和 NSQ，并通過本地文件序列化保存發(fā)送失敗的 event body 并重試。通過插件的方式，用戶在 flume 中的配置文件中指定 NSQ 作為 flume 的下游。

為了支撐更多樣的業(yè)務(wù)需求，有贊 NSQ 還在繼續(xù)完善和豐富更多新特性， 這些特性包括 NSQ 本身的特性開發(fā)，也包括基于 NSQ 做的外部擴(kuò)展系統(tǒng)的開發(fā)。未來的一段時(shí)間，我們計(jì)劃增加如下值得期待的重要特性。

目前有贊有大量的 topic 都部署在一個(gè)大的集群，受益于 golang 的goroutine模型，每個(gè)topic基本都是獨(dú)立的處理，互相直接影響不大, 但是碰到一些數(shù)據(jù)量大的情況, 還是會(huì)對其他topic造成一定的影響，特別是一些網(wǎng)絡(luò)流量非常大的 topic，為了降低這種topic流量影響，我們需要限制一些topic的流量上限， 對整個(gè)集群的穩(wěn)定性提供保障。 設(shè)計(jì)方案上， 我們計(jì)劃使用業(yè)界常用的令牌桶方案。

目前的 NSQ 還是沿用每條消息 ack 的模式， 保持兼容特性。 性能上雖然滿足目前以及未來一段時(shí)間的業(yè)務(wù)需求，但是還有改進(jìn)的空間。特別是在某些網(wǎng)絡(luò)延遲較高的場景下，批量訂閱可以大大提高吞吐量。批量訂閱將會(huì)支持一次消費(fèi)一組消息并且可以一次性 ack 一組消息，從而減少一定的網(wǎng)絡(luò)開銷。

原版的 NSQ 已經(jīng)支持一部分的安全審計(jì)功能, 包括使用安全鏈接以及使用驗(yàn)證服務(wù)器，我們后面將會(huì)針對 topic 的生產(chǎn)和 channel 的一些操作提供獨(dú)立的安全驗(yàn)證服務(wù)，并做好審計(jì)日志，防范一些安全問題。另外針對 nsqadmin 也會(huì)打通內(nèi)部的統(tǒng)一登錄驗(yàn)證，針對性的限制業(yè)務(wù)的一些危險(xiǎn)操作。

微服務(wù)拆分的痛點(diǎn)就是多個(gè)系統(tǒng)之間的一致性保證問題，因此急需一個(gè)統(tǒng)一的框架能解決此類問題。分布式事務(wù)協(xié)調(diào)器將會(huì)是構(gòu)建在 NSQ 基礎(chǔ)之上的一個(gè)重要產(chǎn)品, 該產(chǎn)品將會(huì)充分利用 NSQ 的一些特性去解決業(yè)務(wù)的痛點(diǎn)。

雖然目前已經(jīng)有支持基于消息擴(kuò)展頭進(jìn)行初步過濾的功能，但是也有些業(yè)務(wù)需求非常定制化，需要更加復(fù)雜的過濾規(guī)則，這種情況為了避免給 NSQ 核心代碼帶來影響，我們也計(jì)劃在 NSQ 之上構(gòu)建一個(gè)更加復(fù)雜的過濾系統(tǒng)去做和業(yè)務(wù)耦合的事情，避免給 NSQ 注入過多的業(yè)務(wù)耦合功能.

本文中，先展示了有贊中間件在 NSQ 中引入的支持拓展的消息格式，并通過 3 個(gè)業(yè)務(wù)場景來展示新的消息格式的玩法。之后的部分介紹了圍繞自研版本 NSQ 開發(fā)的拓展工具，包括了用于遷移的代理，以及可以將 NSQ 與 spark 和 flume 進(jìn)行集成的拓展。最后對于未來計(jì)劃進(jìn)行了介紹，展望了部分計(jì)劃中的新特性。

[1]NSQ spark consumer: https://github.com/youzan/spa...

[2]NSQ flume sink: https://github.com/DoraALin/f...

點(diǎn)擊鏈接，可直達(dá)《How we redesigned the NSQ》系列所有文章：
https://tech.youzan.com/tag/p...