EMS

Extend MySQL Stream；一種基于 MySQL 實(shí)現(xiàn)的 stream 隊列。

功能

集群消費(fèi)、廣播消費(fèi)

自動重試、死信隊列

快速重置消息位點(diǎn)，快速回放消息,快速查詢消息

消息可基于磁盤積壓、消息可快速清理

監(jiān)控 group 積壓，topic 消息量排行，消息鏈路追蹤，消息消費(fèi)超時告警；

讀寫性能 1200-3000 QPS 左右

寫入設(shè)計

msg id 就是 topic 維度的自增 id，可對多個 topic 并發(fā)寫入

針對一個 topic，需要有物理 physics offset， 每次寫入，topic 維度的 physics_offset 自增加一

如果使用 redis 自增特性實(shí)現(xiàn), 為每個 topic 配置一個自增 key, 則可以避免加鎖.

redis 實(shí)現(xiàn)雖然性能好, 如為配置aof,宕機(jī)則可能導(dǎo)致丟失數(shù)據(jù), 此時,會出現(xiàn) offset 重復(fù)異常, 過一會隨著繼續(xù)自增, 也就恢復(fù)了.

寫入需要上鎖嗎? 看怎么寫, 如果使用非原子的形式自增 id,比如數(shù)據(jù)的的方式,先查出最大 id,再加一,那么必須加鎖

topic 維度的自增 id 如果使用 mysql 實(shí)現(xiàn)， 性能不堪受辱，因此，此處使用 redis 自增實(shí)現(xiàn)（可配置為 mysql 實(shí)現(xiàn)）；

經(jīng)過測試，筆記本電腦，單 topic 20 并發(fā)寫入，qps 在 1000-1500 左右（local mysql & local redis），基本滿足業(yè)務(wù)需求。

考慮到高可用性和業(yè)務(wù)場景，此處無法使用批量插入

所有的 topic 和 msg 都寫入的這一張表中，表數(shù)據(jù)定時清理，消費(fèi)完的消息，可提前刪除。

注意，本方案寫入性能瓶頸是 MySQL Server 的性能瓶頸。

讀取設(shè)計

假設(shè)針對一個 topic，只有一個 consumer，只需循環(huán)讀取，然后更新 offset 即可。

但結(jié)合實(shí)際業(yè)務(wù)場景，這種基本不存在，所以，忽略這種場景。

通常，一個 topic 有多個 consumer group（簡稱 tg）， 一個 consumer group 有多個 client（jvm or thread）

如果一個 topic + group（簡稱 tg），有多個 consumer，每個 consumer 有多個線程，讀取和更新 offset 則會有并發(fā)問題, 如下圖。

這個 client id，我們將其設(shè)計為，ip + pid + uuid + thread id；

ip 和 pid 可幫助我們追溯問題

uuid 簡單防重復(fù)

thread id，一種性能優(yōu)化，下面細(xì)說。

結(jié)合實(shí)際業(yè)務(wù)場景，且遵循 simple is better 原則，讀取時，使用上鎖的方式解決并發(fā)問題。鎖的粒度就是 tg

考慮到要實(shí)現(xiàn)基本的順序讀取和防止重復(fù)消費(fèi)，多線程并發(fā)時，我們應(yīng)當(dāng)實(shí)現(xiàn)基于自增的形式讀取 msg；每個 clientid 讀取消息后，都會記錄一個簡單的log，并在 tg 維度增加一個 max offset

每次讀取消息時，每個 client 都需要去檢查當(dāng)前想要讀取的 tg 是否已經(jīng)有【其他 client】在操作 max offset。即，我們將鎖的粒度縮小到了 max offset；

對這個 tg 維度的 max offset + n

批量插入這個 tg + clientid offset log，表明這個消息被這個 clientid 讀取了，同時也間接更新了 max offset（order by offset）

釋放鎖

拉取剛剛讀取的 msg id list 里面的消息體

交給業(yè)務(wù)處理消息

整體原則是，一個 t + g 的 max offset，同時只能有一個 thread 操作（寫和更新）

如果有其他人在讀取，則阻塞

如果沒有其他人在讀取，則鎖住這個 tg， 并批量拉取一定數(shù)量的消息 id，

ack

對于集群消息，如何保證在斷電情況下，消息不丟失,使用數(shù)據(jù)庫存儲消息, 寫入即不會丟失, 但消費(fèi)時, 如果剛剛讀進(jìn)內(nèi)存就立刻宕機(jī),則需要在重啟時恢復(fù)消息.

每個 client get 到消息后，都需要記錄 msg pid，consumer group，state（start、done，retry）為 start 狀態(tài)

ack success，將 log update 為 done 狀態(tài)

ack fail 后，將 log update 為 retry 狀態(tài)，同時將消息存入重試隊列

如果 client 還存活，超過 1 分鐘(可配)，則將其撈出，放進(jìn)重試隊列，并在 10s 進(jìn)行第一次重試

如果 client 還存活，則立刻將其撈出，放進(jìn)重試隊列，并在 10s 進(jìn)行第一次重試

這里需要上鎖嗎？其實(shí)是不需要的，因?yàn)楦碌木S度是 client id 的 log，不存在并發(fā)更新. 這里更新狀態(tài)是表示這些消息已經(jīng)處理結(jié)束了，否則無法判定宕機(jī)場景。

對于 start 狀態(tài)的消息，定時任務(wù)會去檢查

ack 是批量的，ack 失敗，僅會導(dǎo)致重復(fù)消費(fèi)。

廣播消息

是否為廣播消息由 topic 確定

廣播消息不需要上鎖，每一個訂閱該 topic 的 client 都會讀取到該消息

廣播消息不需要 ack，不需要記錄成功或失敗或重試，僅需要內(nèi)存里記錄 offset

推薦盡可能使用集群模式，使用集群模式模擬廣播模式

client id

只有 consumer 需要 client id

client id 由 ip pid uuid + thread id 組成, 可溯源.

client id 需要續(xù)約（5s），如果機(jī)器宕機(jī)，則會被自動清除，且他的 start 狀態(tài)的消息會進(jìn)入重試隊列，交給同 group 的其他 client

client id 可以自己主動注銷，注銷前，自己內(nèi)存的消息應(yīng)當(dāng)被優(yōu)雅消費(fèi)結(jié)束，一般來講，kill -15 的 jvm 都會主動注銷 client id；

核心表設(shè)計

topic 表：記錄topic 元信息

group 表：記錄 group 訂閱元信息

msg 表：msg總表，記錄寫入的信息，包含 body 和 topic 維度的自增 offset，類似 rocketmq commit log

該表會被多個 consumer 消費(fèi)的消息

該表會被定制刪除過期數(shù)據(jù)

retry msg 表，消費(fèi)失敗、超時的消息，會進(jìn)入該表，并按階梯定時消費(fèi)

dead msg 表，消費(fèi)重試 16（any config） 次的消息，會進(jìn)入該表

topic_group_log 表：記錄 consumer group client 的 msg 消費(fèi)記錄，包含 state（start、done，retry） 字段，可 ack

該表的記錄行數(shù)會非常多，單行數(shù)據(jù)較少，可自動刪除 done 的記錄

鎖

如上文所說，由于本方案未采用常見的多 queue 和多 partition 的設(shè)計，因此瓶頸在于上圖提到的分布式鎖的設(shè)計上，具體鏈路為 consumer group client 在集群消費(fèi)時， 為了讓并發(fā)讀取的 thread 拉取到的消息盡可能準(zhǔn)確，使用上鎖的方式來實(shí)現(xiàn)。

總體看下來，可以簡單理解為，ems 失去了性能，卻擁有了所有。

審核編輯：劉清