前提

生產者給kakfa投遞的(de)消(xiao)息(xi)，在同一(yi)個topic下不同的(de)Partition中，不同的(de)Partition之間(jian)的(de)消(xiao)息(xi)是無(wu)法(fa)保(bao)證順(shun)(shun)序的(de)。但(dan)有些場(chang)景(jing)，業務(wu)需(xu)要處(chu)理順(shun)(shun)序消(xiao)息(xi)。怎么(me)保(bao)證kakfa消(xiao)息(xi)的(de)順(shun)(shun)序性，以致最(zui)終(zhong)消(xiao)費(fei)者能(neng)順(shun)(shun)序消(xiao)費(fei)消(xiao)息(xi)呢

消息有序性分類

• 全局有序：一個Topic下的所有消息都需要按照生產順序消費。
• 局部有序：一個Topic下的消息，只需要滿足同一業務字段的要按照生產順序消費。例如：Topic消息是訂單的流水表，包含訂單orderId，業務要求同一個orderId的消息需要按照生產順序進行消費。

全局有序

由(you)于Kafka的(de)(de)一個(ge)Topic可以分(fen)為了多個(ge)Partition，Producer發送消(xiao)息的(de)(de)時候(hou)，是分(fen)散(san)在不(bu)同 Partition的(de)(de)。當(dang)Producer按(an)順(shun)序(xu)發消(xiao)息給(gei)Broker，但進(jin)入Kafka之后，這些消(xiao)息就不(bu)一定進(jin)到哪個(ge)Partition，會(hui)導(dao)致順(shun)序(xu)是亂的(de)(de)。因此要滿足全局有(you)序(xu)，需要1個(ge)Topic只能對應(ying)(ying)1個(ge)Partition。而(er)且(qie)對應(ying)(ying)的(de)(de)consumer也要使用單線程(cheng)或者(zhe)保(bao)證消(xiao)費順(shun)序(xu)的(de)(de)線程(cheng)模型(xing).

局部有序

要(yao)滿足局部有序，只需要(yao)在(zai)發消(xiao)(xiao)(xiao)息(xi)的時候指定Partition Key，Kafka對其進行Hash計算(suan)，根據計算(suan)結果決定放入(ru)哪個Partition。這樣(yang)Partition Key相(xiang)同(tong)(tong)(tong)的消(xiao)(xiao)(xiao)息(xi)會放在(zai)同(tong)(tong)(tong)一(yi)個Partition。確(que)保將相(xiang)關(guan)的消(xiao)(xiao)(xiao)息(xi)發送到同(tong)(tong)(tong)一(yi)個分區(qu)。Kafka 保證在(zai)單個分區(qu)內消(xiao)(xiao)(xiao)息(xi)的順序。因(yin)此(ci)(ci)，如果消(xiao)(xiao)(xiao)息(xi)的順序性對業務非常重要(yao)，可以將相(xiang)關(guan)的消(xiao)(xiao)(xiao)息(xi)通(tong)過相(xiang)同(tong)(tong)(tong)的分區(qu)鍵（如用戶(hu) ID 或訂單 ID）發送至同(tong)(tong)(tong)一(yi)分區(qu)。此(ci)(ci)時，Partition的數量仍(reng)然可以設置(zhi)多個，提升Topic的整體吞吐量。

提升消費效率

在不增加(jia)partition數量的(de)情況(kuang)下想提高消(xiao)(xiao)費速度，可以考慮以上局部(bu)有(you)序的(de)場(chang)景下再(zai)次hash唯一標識（例(li)(li)如(ru)訂(ding)單orderId）到(dao)不同的(de)線程(cheng)上，多個(ge)(ge)消(xiao)(xiao)費者(zhe)線程(cheng)并發處(chu)理消(xiao)(xiao)息（依舊可以保(bao)證(zheng)局部(bu)有(you)序）。說到(dao)底(di)還是一個(ge)(ge)實例(li)(li)線程(cheng)消(xiao)(xiao)費一個(ge)(ge)partition消(xiao)(xiao)息

消息重試對順序消息的影響

對于(yu)一個(ge)有著先后順序(xu)(xu)的(de)消息A、B，正常情(qing)況(kuang)下(xia)應該是A先發送(song)(song)(song)完成(cheng)(cheng)(cheng)后再發送(song)(song)(song)B，但是在(zai)異常情(qing)況(kuang)下(xia)，在(zai)A發送(song)(song)(song)失敗(bai)的(de)情(qing)況(kuang)下(xia)，B發送(song)(song)(song)成(cheng)(cheng)(cheng)功(gong)(gong)，而A由(you)于(yu)重試機制在(zai)B發送(song)(song)(song)完成(cheng)(cheng)(cheng)之后重試發送(song)(song)(song)成(cheng)(cheng)(cheng)功(gong)(gong)了(le)(le)。這時(shi)對于(yu)本身(shen)順序(xu)(xu)為(wei)AB的(de)消息順序(xu)(xu)變成(cheng)(cheng)(cheng)了(le)(le)BA。

針對這種問題，嚴格的順序消費還需要max.in.flight.requests.per.connection參數的支持。

該參(can)數指定了生產者在收到(dao)服務器響(xiang)應之前可以發送多少個消息。它的(de)值越高(gao)，就會占用越多的(de)內存，同(tong)時也會提升吞吐(tu)量。把它設為1就可以保(bao)證消息是(shi)按照發送的(de)順序寫入服務器的(de)。

此外，對于某些業務場景，設置max.in.flight.requests.per.connection=1會嚴重降低(di)吞吐(tu)量，如果放棄(qi)使用這(zhe)種同步重試機制，則(ze)可(ke)以考慮(lv)在消(xiao)費端增加失(shi)敗標(biao)記(ji)(ji)的(de)(de)記(ji)(ji)錄，然后(hou)用定時任(ren)務輪詢去重試這(zhe)些失(shi)敗的(de)(de)消(xiao)息并做好監控報(bao)警。