一文理解Kafka如何保證消息順序性

要想實(shí)現(xiàn)消息有序，需要從Producer和Consumer兩方面來(lái)考慮。

如果對(duì)Kafka不了解的話，可以先看這篇博客《一文快速了解Kafka》。

針對(duì)消息有序的業(yè)務(wù)需求，還分為全局有序和局部有序。

• 全局有序：一個(gè)Topic下的所有消息都需要按照生產(chǎn)順序消費(fèi)。

• 局部有序：一個(gè)Topic下的消息，只需要滿足同一業(yè)務(wù)字段的要按照生產(chǎn)順序消費(fèi)。例如：Topic消息是訂單的流水表，包含訂單orderId，業(yè)務(wù)要求同一個(gè)orderId的消息需要按照生產(chǎn)順序進(jìn)行消費(fèi)。

全局有序

由于Kafka的一個(gè)Topic可以分為了多個(gè)Partition，Producer發(fā)送消息的時(shí)候，是分散在不同 Partition的。當(dāng)Producer按順序發(fā)消息給Broker，但進(jìn)入Kafka之后，這些消息就不一定進(jìn)到哪個(gè)Partition，會(huì)導(dǎo)致順序是亂的。

因此要滿足全局有序，需要1個(gè)Topic只能對(duì)應(yīng)1個(gè)Partition。

而且對(duì)應(yīng)的consumer也要使用單線程或者保證消費(fèi)順序的線程模型，否則會(huì)出現(xiàn)下圖所示，消費(fèi)端造成的消費(fèi)亂序。

局部有序

要滿足局部有序，只需要在發(fā)消息的時(shí)候指定Partition Key，Kafka對(duì)其進(jìn)行Hash計(jì)算，根據(jù)計(jì)算結(jié)果決定放入哪個(gè)Partition。這樣Partition Key相同的消息會(huì)放在同一個(gè)Partition。此時(shí)，Partition的數(shù)量仍然可以設(shè)置多個(gè)，提升Topic的整體吞吐量。

如下圖所示，在不增加partition數(shù)量的情況下想提高消費(fèi)速度，可以考慮再次hash唯一標(biāo)識(shí)（例如訂單orderId）到不同的線程上，多個(gè)消費(fèi)者線程并發(fā)處理消息（依舊可以保證局部有序）。

消息重試對(duì)順序消息的影響

對(duì)于一個(gè)有著先后順序的消息A、B，正常情況下應(yīng)該是A先發(fā)送完成后再發(fā)送B，但是在異常情況下，在A發(fā)送失敗的情況下，B發(fā)送成功，而A由于重試機(jī)制在B發(fā)送完成之后重試發(fā)送成功了。這時(shí)對(duì)于本身順序?yàn)锳B的消息順序變成了BA。

針對(duì)這種問(wèn)題，嚴(yán)格的順序消費(fèi)還需要max.in.flight.requests.per.connection參數(shù)的支持。

該參數(shù)指定了生產(chǎn)者在收到服務(wù)器響應(yīng)之前可以發(fā)送多少個(gè)消息。它的值越高，就會(huì)占用越多的內(nèi)存，同時(shí)也會(huì)提升吞吐量。把它設(shè)為1就可以保證消息是按照發(fā)送的順序?qū)懭敕?wù)器的。

此外，對(duì)于某些業(yè)務(wù)場(chǎng)景，設(shè)置max.in.flight.requests.per.connection=1會(huì)嚴(yán)重降低吞吐量，如果放棄使用這種同步重試機(jī)制，則可以考慮在消費(fèi)端增加失敗標(biāo)記的記錄，然后用定時(shí)任務(wù)輪詢?nèi)ブ卦囘@些失敗的消息并做好監(jiān)控報(bào)警。