存在的意義
業(yè)務(wù)直接去接kafka或者其他的mq，需要自己處理rebalance場(chǎng)景下的各種極端case，以及一些安全瞪讼，網(wǎng)絡(luò)分區(qū)問題。
SDK直連kafka的話粹断，會(huì)面臨消費(fèi)者數(shù)量不能超過partition數(shù)量的限制符欠。如果有proxy做代理，可以解決這個(gè)問題瓶埋。而且可以降低rebalance的概率希柿。
重復(fù)消費(fèi)問題：重復(fù)消費(fèi)和rebalance其實(shí)沒有關(guān)系，和重連有關(guān)系悬赏，只要重連都有可能重復(fù)消費(fèi)狡汉。只要proxy不發(fā)版重啟，理論上就不會(huì)有重復(fù)消費(fèi)闽颇。
消息投遞SLA
At Least Once
At Least Once是絕大多數(shù)消息服務(wù)系統(tǒng)的SLA(服務(wù)水平承諾)盾戴，即指所有投遞成功的消息將會(huì)被至少成功消費(fèi)一次。對(duì)于MSM來說兵多，普通消息/順序消息/延遲消息類型均符合這一SLA尖啡，其含義有三點(diǎn)： 1）生產(chǎn)者需要成功投遞消息。2）消費(fèi)者需要使用ack機(jī)制來確保消息消費(fèi)的成功剩膘。3）在某些情況下衅斩，消費(fèi)者收到的消息是可能產(chǎn)生重復(fù)的，比如怠褐，proxy實(shí)例重啟畏梆，消費(fèi)者client非優(yōu)雅退出，后端provider重啟等等。消費(fèi)者需要有一定的容錯(cuò)機(jī)制來避免少量重復(fù)消息的影響奠涌。
At Most Once
At Most Once是另一種SLA宪巨，指所有成功投遞的消息將會(huì)被至多消費(fèi)一次，采用At Most Once的系統(tǒng)將不會(huì)有重復(fù)的消費(fèi)溜畅，但是可能會(huì)丟失消息捏卓，其吞吐量相比At Least Once類型要高一個(gè)數(shù)量級(jí)。目前內(nèi)存型消息隊(duì)列均為At Most Once慈格，如nats怠晴，nsq等。
Exact Once
Exact Once是最嚴(yán)格的一種SLA浴捆，即所有被成功投遞的消息都將被恰好消費(fèi)一次蒜田，采用Exact Once的系統(tǒng)既不會(huì)有重復(fù)消費(fèi)也不會(huì)有消息丟失，只有投遞成功或失敗汤功，其吞吐量顯著低于前兩種SLA物邑。一般只有事務(wù)消息類型會(huì)采用exact once的SLA溜哮。
常見消息類型
* **普通消息**滔金。遵循隊(duì)列先進(jìn)先出，同一個(gè)consumer實(shí)例消費(fèi)消息有序茂嗓，不同的consumer實(shí)例之間消費(fèi)消息無序餐茵，除非特殊場(chǎng)景，正常的生產(chǎn)和消費(fèi)都應(yīng)該使用普通消息述吸。
* **順序消息**忿族。topic的某個(gè)subscription同一時(shí)間只允許一個(gè)consumer實(shí)例消費(fèi)，消息嚴(yán)格有序蝌矛。
* **延遲消息**道批。生產(chǎn)的消息并不第一時(shí)間可見，而是在規(guī)定的時(shí)間之后可以被consumer消費(fèi)入撒，但是并不保證一定是在規(guī)定的時(shí)間被消費(fèi)隆豹。
* **重投隊(duì)列**。消費(fèi)失敗的消息茅逮，用戶可以配置進(jìn)入重投隊(duì)列進(jìn)行一定次數(shù)的重投璃赡，重投超過次數(shù)之后將進(jìn)入死信隊(duì)列。
* **死信隊(duì)列**献雅。用戶可以在申請(qǐng)topic時(shí)同時(shí)申請(qǐng)一個(gè)死信隊(duì)列碉考，任何消費(fèi)失敗(Nack)的消息都會(huì)進(jìn)入死信隊(duì)列，用戶可以離線消費(fèi)死信隊(duì)列的消息挺身。
* **事務(wù)消息**侯谁。生產(chǎn)者投遞消息之后，執(zhí)行事務(wù)處理邏輯，根據(jù)處理的結(jié)果執(zhí)行Commit或者Rollback墙贱，只有執(zhí)行成功的消息才能被消費(fèi)者消費(fèi)到（兩階段提交）技扼，參考流程： [https://rocketmq.apache.org/docs/4.x/producer/06message5/](https://rocketmq.apache.org/docs/4.x/producer/06message5/)。
特性
容災(zāi)和可用性
無狀態(tài)proxy嫩痰。目前proxy服務(wù)是無狀態(tài)剿吻，理論上可以無限水平擴(kuò)容，其任何一個(gè)節(jié)點(diǎn)都是可替換的串纺。
存儲(chǔ)provider丽旅。provider主要為云消息隊(duì)列產(chǎn)品，依賴云產(chǎn)品的可用性纺棺。國(guó)內(nèi)普通消息類型使用的是aliyun kafka榄笙，延遲消息類型使用的是rocketmq，海外延遲消息類型使用的是sqs祷蝌。
topic遷移茅撞。支持topic從一個(gè)provider實(shí)例(zone)無縫遷移到另一個(gè)provider實(shí)例(zone)，支持topic在發(fā)生zone級(jí)別的容災(zāi)時(shí)在有損條件下直接遷移巨朦，從而實(shí)現(xiàn)異地多活米丘。
連接管理。支持不同業(yè)務(wù)的proxy集群劃分糊啡，做到更嚴(yán)格的租戶隔離和資源獨(dú)享拄查。
使用維度：
跨區(qū)生產(chǎn)消費(fèi)，國(guó)內(nèi)生產(chǎn)海外消費(fèi)
自定義監(jiān)控棚蓄，消息軌跡堕扶，限流
安全性，租戶管理/隔離
批量處理梭依，順序提交
會(huì)在sdk和proxy都維護(hù)一個(gè)滑動(dòng)窗口稍算，隨著窗口移動(dòng)提交。如果有一個(gè)一直沒提交役拴，則窗口將停止移動(dòng)糊探。
proxy和sdk的關(guān)系
sdk可以理解為業(yè)務(wù)的消費(fèi)者，sdk與proxy通過grpc的stream進(jìn)行連接扎狱，proxy為每個(gè)stream維護(hù)一個(gè)滑動(dòng)窗口侧到。
proxy的數(shù)量根據(jù)整個(gè)公司的業(yè)務(wù)情況決定，大部分場(chǎng)景下是不需要改變的淤击。申請(qǐng)的kafka的topic匠抗，比如有6個(gè)partition，但是有100個(gè)proxy污抬，會(huì)隨機(jī)連到6個(gè)proxy上汞贸。然后有一個(gè)地方記錄topic和對(duì)應(yīng)proxy實(shí)例的列表绳军，sdk連接topic的時(shí)候，就會(huì)找到實(shí)例列表矢腻，然后建立stream门驾。
如果需要提高消費(fèi)速度，一個(gè)業(yè)務(wù)進(jìn)程多柑，可以開多消費(fèi)者奶是，建立多個(gè)stream，每個(gè)stream有自己的滑動(dòng)窗口竣灌，從而提高并發(fā)量聂沙。
注意：并發(fā)量越高，重連的時(shí)候初嘹，可能重復(fù)消費(fèi)的消息就越多及汉。
消息中間件代理還能做哪些事
消息中間件代理可以在業(yè)務(wù)層和消息中間件（如 Kafka、RabbitMQ屯烦、ActiveMQ 等）之間提供附加功能和服務(wù)坷随。以下是一些消息中間件代理可以做的事情：
1. **消息轉(zhuǎn)換**：代理可以將消息從一種格式轉(zhuǎn)換為另一種格式，以滿足業(yè)務(wù)層的需求驻龟。例如温眉，代理可以將 JSON 消息轉(zhuǎn)換為 Protocol Buffers 格式或 Avro 格式。
2. **消息過濾**：代理可以根據(jù)業(yè)務(wù)層的需求過濾消息迅脐。例如芍殖，代理可以只將包含某個(gè)關(guān)鍵詞的消息傳遞給業(yè)務(wù)層，以減小業(yè)務(wù)層處理無關(guān)消息的負(fù)擔(dān)谴蔑。
3. **消息路由**：代理可以根據(jù)消息內(nèi)容或其他規(guī)則將消息路由到不同的消費(fèi)者或業(yè)務(wù)組件。這可以幫助您實(shí)現(xiàn)更靈活的消息處理架構(gòu)龟梦，例如將高優(yōu)先級(jí)消息發(fā)送到高性能的處理組件隐锭。
4. **消息緩存**：代理可以緩存消息，以便在消費(fèi)者出現(xiàn)問題時(shí)重新發(fā)送消息计贰。這有助于提高系統(tǒng)的可靠性钦睡，確保消息不會(huì)因?yàn)榕R時(shí)故障而丟失。
5. **消息重試和死信隊(duì)列**：代理可以實(shí)現(xiàn)消息重試策略躁倒，當(dāng)業(yè)務(wù)層處理失敗時(shí)荞怒，代理可以在一段時(shí)間后重新發(fā)送消息。同時(shí)秧秉，代理還可以實(shí)現(xiàn)死信隊(duì)列功能褐桌，將連續(xù)失敗的消息移入死信隊(duì)列，以便后續(xù)處理和分析象迎。
6. **負(fù)載均衡**：代理可以實(shí)現(xiàn)負(fù)載均衡策略荧嵌，根據(jù)消費(fèi)者的處理能力動(dòng)態(tài)調(diào)整消息的分發(fā)呛踊。這有助于確保系統(tǒng)資源的充分利用，提高處理性能啦撮。
7. **監(jiān)控和度量**：代理可以收集關(guān)于消息處理的度量和統(tǒng)計(jì)信息谭网，例如消息處理速度、失敗率赃春、延遲等愉择。這些信息可以幫助您監(jiān)控系統(tǒng)的性能和健康狀況，及時(shí)發(fā)現(xiàn)和解決問題织中。
8. **安全和認(rèn)證**：代理可以實(shí)現(xiàn)消息的加密和解密薄辅，以保護(hù)消息傳輸過程中的數(shù)據(jù)安全。此外抠璃，代理還可以實(shí)現(xiàn)消費(fèi)者和生產(chǎn)者的身份驗(yàn)證和授權(quán)站楚，以確保只有合法的用戶可以訪問消息。
通過實(shí)現(xiàn)這些功能搏嗡，消息中間件代理可以幫助您構(gòu)建更強(qiáng)大窿春、靈活且可靠的消息處理系統(tǒng)。代理可以讓業(yè)務(wù)層更關(guān)注業(yè)務(wù)邏輯采盒，而無需處理消息中間件的底層細(xì)節(jié)旧乞。同時(shí)，代理還可以提供一組豐富的工具和服務(wù)磅氨，助力您更好地監(jiān)控尺栖、調(diào)試和優(yōu)化您的系統(tǒng)。
生產(chǎn)環(huán)境消息隊(duì)列使用注意事項(xiàng)
生產(chǎn)者優(yōu)雅關(guān)機(jī)：很多生產(chǎn)者的客戶端都是異步的烦租，本地有緩存延赌，生產(chǎn)接口返回成功，并不代表消息已經(jīng)寫到遠(yuǎn)端了【好的sdk叉橱，多線程提交會(huì)等寫到遠(yuǎn)端以后再返回】挫以。關(guān)機(jī)前，先停止生產(chǎn)窃祝，在flush一下掐松，然后關(guān)機(jī)。
消費(fèi)者優(yōu)雅關(guān)機(jī)：先停止消費(fèi)粪小，然后commit已經(jīng)處理的大磺，之后關(guān)機(jī)。盡量減少重復(fù)消費(fèi)的消息數(shù)量
滑動(dòng)窗口實(shí)現(xiàn)
消息中間件proxy在面多對(duì)個(gè)客戶端連接的時(shí)候如何做滑動(dòng)窗口
在消息中間件代理面對(duì)多個(gè)客戶端連接的情況下探膊，可以使用滑動(dòng)窗口算法來實(shí)現(xiàn)流量控制和負(fù)載均衡杠愧。滑動(dòng)窗口算法可以確保代理不會(huì)因?yàn)閱蝹€(gè)客戶端的延遲或擁塞而影響其他客戶端突想。以下是使用滑動(dòng)窗口算法的實(shí)現(xiàn)步驟：
1. **為每個(gè)客戶端分配窗口**：給每個(gè)連接的客戶端分配一個(gè)固定大小的窗口殴蹄。這個(gè)窗口表示代理可以在任何時(shí)候向該客戶端發(fā)送的最大消息數(shù)量究抓。例如，如果窗口大小為 5袭灯，代理最多可以在等待客戶端響應(yīng)之前發(fā)送 5 條消息刺下。
2. **記錄已發(fā)送消息**：對(duì)于每個(gè)客戶端，代理需要記錄已發(fā)送但尚未收到確認(rèn)的消息稽荧。這可以通過一個(gè)隊(duì)列或列表來實(shí)現(xiàn)橘茉。同時(shí)，代理還需要記錄發(fā)送的消息數(shù)量姨丈。
3. **發(fā)送消息**：當(dāng)向客戶端發(fā)送消息時(shí)畅卓，代理首先檢查客戶端的滑動(dòng)窗口是否已滿（即已發(fā)送但未確認(rèn)的消息數(shù)量等于窗口大小）蟋恬。如果窗口未滿翁潘，則代理可以發(fā)送新消息，并將新消息添加到已發(fā)送消息記錄中歼争。
4. **處理確認(rèn)**：當(dāng)代理收到客戶端的消息確認(rèn)時(shí)（例如拜马，客戶端成功處理了一條消息），代理需要將對(duì)應(yīng)的消息從已發(fā)送消息記錄中移除沐绒，并將滑動(dòng)窗口向前移動(dòng)（即增加可發(fā)送的消息數(shù)量）俩莽。
5. **負(fù)載均衡**：代理可以根據(jù)每個(gè)客戶端的滑動(dòng)窗口使用情況來實(shí)現(xiàn)負(fù)載均衡。例如乔遮，代理可以優(yōu)先向擁有更多可用窗口空間的客戶端發(fā)送消息扮超，或者根據(jù)窗口使用率來分配消息。這可以確保代理根據(jù)客戶端的處理能力來分發(fā)消息蹋肮，從而實(shí)現(xiàn)更有效的負(fù)載均衡出刷。
6. **超時(shí)和重試**：為了處理丟失的確認(rèn)或失敗的消息傳遞，代理可以引入超時(shí)和重試機(jī)制括尸。當(dāng)已發(fā)送的消息在規(guī)定時(shí)間內(nèi)未收到確認(rèn)時(shí)巷蚪，代理可以將消息重發(fā)，并更新滑動(dòng)窗口濒翻。同時(shí)，代理還可以實(shí)現(xiàn)一個(gè)最大重試次數(shù)限制啦膜，以避免無限制的重發(fā)導(dǎo)致資源耗盡有送。
通過使用滑動(dòng)窗口算法，您可以在消息中間件代理中實(shí)現(xiàn)更有效的流量控制和負(fù)載均衡僧家。這可以確保代理在面對(duì)多個(gè)客戶端連接時(shí)能夠平滑地分發(fā)消息雀摘，降低擁塞和延遲，從而提高整個(gè)系統(tǒng)的性能和可靠性八拱。
kafka提交的細(xì)節(jié)
kafka-go在開啟手動(dòng)commit的情況下阵赠，如果老消息如果在沒提交涯塔，去提交新消息會(huì)有什么表現(xiàn)?
在 Kafka-go 中，手動(dòng)提交偏移量時(shí)清蚀，提交的是當(dāng)前消費(fèi)者已經(jīng)處理完畢的最新偏移量匕荸。當(dāng)老消息尚未提交時(shí)，手動(dòng)提交新消息實(shí)際上會(huì)導(dǎo)致將消費(fèi)者的偏移量更新為新消息的偏移量枷邪。這意味著在 Kafka 中榛搔，老消息以及在它之前的所有消息都被認(rèn)為已經(jīng)處理并提交了。
處理這種情況時(shí)东揣，需要注意以下幾點(diǎn)：
1. 數(shù)據(jù)一致性：如果手動(dòng)提交新消息導(dǎo)致了老消息的丟失志衣，您需要確保這不會(huì)對(duì)您的應(yīng)用程序產(chǎn)生負(fù)面影響领迈。在某些情況下，您可能需要修改消費(fèi)者的邏輯以確保在提交新消息之前處理并提交所有舊消息。
2. 消息處理順序：Kafka 中的消息是按分區(qū)內(nèi)的順序進(jìn)行處理的借嗽，但是當(dāng)分區(qū)之間并行處理時(shí)，消息的全局順序可能會(huì)受到影響澎灸。如果您的應(yīng)用程序?qū)樞蛴幸笱肚敲丛谔幚硇孪⒅埃?qǐng)確保已經(jīng)處理并提交所有舊消息运沦。
3. 消息重復(fù)處理：由于 Kafka-go 在開啟手動(dòng)提交模式下可能會(huì)在錯(cuò)誤情況下多次處理相同的消息泵额，所以應(yīng)在消費(fèi)者中處理這種情況。例如携添，使用冪等操作或在數(shù)據(jù)庫(kù)中維護(hù)一個(gè)已處理消息的列表嫁盲。
在使用 Kafka-go 手動(dòng)提交模式時(shí)，應(yīng)確保在提交新消息之前正確處理并提交所有舊消息烈掠。這可以通過檢查消費(fèi)者的錯(cuò)誤日志羞秤、調(diào)整提交間隔等方式來實(shí)現(xiàn)。同時(shí)左敌，確保有足夠的容錯(cuò)處理機(jī)制以應(yīng)對(duì)可能的消息丟失或重復(fù)處理情況瘾蛋。
proxy實(shí)現(xiàn)細(xì)節(jié)
stream級(jí)別的滑動(dòng)窗口
生產(chǎn)stream不需要窗口，生產(chǎn)stream的窗口應(yīng)該sdk做矫限。proxy收到消息以后哺哼，給sdk會(huì)ack，sdk根據(jù)ack去移動(dòng)窗口控制生產(chǎn)速率叼风∪《【實(shí)際生產(chǎn)中，建議一條一條生產(chǎn)无宿，批量生產(chǎn)容易出錯(cuò)茵汰，且不知道錯(cuò)在哪條，可能會(huì)多會(huì)少】
消費(fèi)stream需要窗口孽鸡，sdk消費(fèi)的消息蹂午，先寫window栏豺，再寫和sdk的stream，window滿了則阻塞豆胸，通過sdk的ack去移動(dòng)窗口奥洼，控制消費(fèi)速率∨渎遥【防止單個(gè)stream把某個(gè)partiton的commit-window占滿溉卓，導(dǎo)致頻繁reset】【partion的commit-window大小=partion支持的stream個(gè)數(shù)*每個(gè)stream的window大小，實(shí)際可以設(shè)置的比這個(gè)小一些搬泥，提高reset的頻率桑寨，設(shè)置的越小，會(huì)重復(fù)消費(fèi)的消息越少忿檩，同樣尉尾，在不穩(wěn)定情況下出發(fā)reset的概率也會(huì)增加】
commit細(xì)節(jié)
如果offset為1的消息被push到客戶端，但是還沒ack燥透，這時(shí)候sdk和proxy的stream斷了沙咏，下次stream重連的時(shí)候，還能拿到這條消息嗎
解決辦法就是proxy維護(hù)partition級(jí)別的commit-window班套，如果window滿了【這個(gè)值的設(shè)置需要根據(jù)業(yè)務(wù)情況調(diào)整】肢藐，則reset一下offset
sdk實(shí)現(xiàn)細(xì)節(jié)
sdk的消費(fèi)stream不需要window，由proxy來進(jìn)行限制吱韭，只需要無腦向proxy拉吆豹，處理完以后進(jìn)行commit即可。
sdk的生產(chǎn)stream可以不用window理盆，逐條進(jìn)行投遞痘煤，返回成功則投遞成功。
如果生產(chǎn)stream非要使用window猿规，可以進(jìn)場(chǎng)并發(fā)生產(chǎn)衷快，批量投遞，proxy進(jìn)行ack的時(shí)候姨俩，需要明確指出自己ack了哪幾條消息蘸拔，然后sdk把生產(chǎn)這幾條消息的阻塞的goroutine給解開』房【要保證投遞線程返回的時(shí)候都伪，該線程投遞的數(shù)據(jù)一定都成功了，如果投遞的線程返回err积担，則一定都是失敗的】【建議逐條投遞，沒必要搞批量猬仁，真要搞批量帝璧，也在業(yè)務(wù)級(jí)別搞先誉，即一個(gè)kafka的msg里面存儲(chǔ)多個(gè)業(yè)務(wù)包，而不是只存儲(chǔ)一個(gè)】
總結(jié)
生產(chǎn)的最佳實(shí)踐就是逐條投遞的烁，等返回褐耳，需要批量的話業(yè)務(wù)層自己做。
消費(fèi)的最佳實(shí)踐是批量并發(fā)消費(fèi)渴庆，proxy維護(hù)commit-window和stream-window铃芦，根據(jù)不同的業(yè)務(wù)，調(diào)整reset-window的大小襟雷，選擇合適的reset策略刃滓。在commit-window卡住的時(shí)候，即時(shí)去reset耸弄。
其中只要commit-window不要stream-window也行咧虎，但是會(huì)導(dǎo)致不同的stream之間處理不均衡，所以最好還是都要一下计呈。
很糙的中間件實(shí)現(xiàn)
一個(gè)stream錯(cuò)誤砰诵，直接關(guān)閉reader進(jìn)行rebalance，commit出錯(cuò)也是一樣捌显。
rebalance時(shí)茁彭，中間件正確的操作
rebalance時(shí)read方法的表現(xiàn)
rebalance時(shí)commit方法的表現(xiàn)
正確的處理姿勢(shì)
rebalance時(shí)，讀阻塞扶歪，不需要額外處理理肺。commit失敗，進(jìn)行一定次數(shù)的重試击罪，還失敗的話就關(guān)閉stream哲嘲，但是不需要關(guān)閉reader，因?yàn)榭赡苣鉩ommit的partiton被分給別人了媳禁。
commit失敗的原因
分區(qū)所有權(quán)變更導(dǎo)致commit失敗處理辦法
commit失敗以后眠副，可以查詢一下當(dāng)前reader的分區(qū)所有權(quán)，如果要提交消息不屬于當(dāng)前reader了竣稽，則關(guān)閉stream囱怕，且消息不需要重新投遞，因?yàn)闀?huì)被其他reader再次消費(fèi)到毫别。
renbalance的stalelize
kafka的rebalance是有一個(gè)stablize階段的娃弓，就是一段時(shí)間之內(nèi)連上來的客戶端都會(huì)參與到rebalance，而且這個(gè)時(shí)候是不會(huì)報(bào)錯(cuò)的岛宦，stablize階段過了之后才會(huì)真正rebalance成功台丛，否則的話任何網(wǎng)絡(luò)抖動(dòng)都會(huì)導(dǎo)致頻繁rebalance，kafka server就沒有穩(wěn)定性了。
很多時(shí)候rebalance的表現(xiàn)就是沒有響應(yīng)挽霉，所以當(dāng)出現(xiàn)超時(shí)的時(shí)候防嗡，客戶端很難確定是rebalance還是真的網(wǎng)絡(luò)不通了。
generation
rebalance之后的consumer是帶generation的侠坎，非本generation的commit會(huì)全部失敗蚁趁。所以理論上說consumer必須全部重新取消息重新commit。但是kafka-go幫我們屏蔽了這個(gè)點(diǎn)实胸，只要partion沒被分走他嫡，之前消費(fèi)的消息也可以commit。
