EMS

Extend MySQL Stream；一種基于 MySQL 實現的 stream 隊列。

功能

集群消費、廣播消費

自動重試、死信隊列

快速重置消息位點，快速回放消息,快速查詢消息

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

監控 group 積壓，topic 消息量排行，消息鏈路追蹤，消息消費超時告警；

讀寫性能 1200-3000 QPS 左右

寫入設計

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

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

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

redis 實現雖然性能好, 如為配置aof,宕機則可能導致丟失數據, 此時,會出現 offset 重復異常, 過一會隨著繼續自增, 也就恢復了.

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

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

經過測試，筆記本電腦，單 topic 20 并發寫入，qps 在 1000-1500 左右（local mysql & local redis），基本滿足業務需求。

考慮到高可用性和業務場景，此處無法使用批量插入

所有的 topic 和 msg 都寫入的這一張表中，表數據定時清理，消費完的消息，可提前刪除。

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

讀取設計

假設針對一個 topic，只有一個 consumer，只需循環讀取，然后更新 offset 即可。

但結合實際業務場景，這種基本不存在，所以，忽略這種場景。

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

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

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

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

uuid 簡單防重復

thread id，一種性能優化，下面細說。

結合實際業務場景，且遵循 simple is better 原則，讀取時，使用上鎖的方式解決并發問題。鎖的粒度就是 tg

考慮到要實現基本的順序讀取和防止重復消費，多線程并發時，我們應當實現基于自增的形式讀取 msg；每個 clientid 讀取消息后，都會記錄一個簡單的log，并在 tg 維度增加一個 max offset

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

對這個 tg 維度的 max offset + n

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

釋放鎖

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

交給業務處理消息

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

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

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

ack

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

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

ack success，將 log update 為 done 狀態

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

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

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

這里需要上鎖嗎？其實是不需要的，因為更新的維度是 client id 的 log，不存在并發更新. 這里更新狀態是表示這些消息已經處理結束了，否則無法判定宕機場景。

對于 start 狀態的消息，定時任務會去檢查

ack 是批量的，ack 失敗，僅會導致重復消費。

廣播消息

是否為廣播消息由 topic 確定

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

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

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

client id

只有 consumer 需要 client id

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

client id 需要續約（5s），如果機器宕機，則會被自動清除，且他的 start 狀態的消息會進入重試隊列，交給同 group 的其他 client

client id 可以自己主動注銷，注銷前，自己內存的消息應當被優雅消費結束，一般來講，kill -15 的 jvm 都會主動注銷 client id；

核心表設計

topic 表：記錄topic 元信息

group 表：記錄 group 訂閱元信息

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

該表會被多個 consumer 消費的消息

該表會被定制刪除過期數據

retry msg 表，消費失敗、超時的消息，會進入該表，并按階梯定時消費

dead msg 表，消費重試 16（any config） 次的消息，會進入該表

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

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

鎖

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

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

審核編輯：劉清