Pulsar源碼徹底解決重復(fù)消費(fèi)問題

更新時(shí)間：2023年05月29日 11:29:53 作者：crossoverJie

這篇文章主要為大家介紹了Pulsar源碼徹底解決重復(fù)消費(fèi)問題，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

背景

最近真是和 Pulsar 杠上了，業(yè)務(wù)團(tuán)隊(duì)反饋說是線上有個(gè)應(yīng)用消息重復(fù)消費(fèi)。

而且在測試環(huán)境是可以穩(wěn)定復(fù)現(xiàn)的，根據(jù)經(jīng)驗(yàn)來看一般能穩(wěn)定復(fù)現(xiàn)的都比較好解決。

定位問題

接著便是定位問題了，根據(jù)之前的經(jīng)驗(yàn)讓業(yè)務(wù)按照這幾種情況先排查一下：

通過排查：1,2可以排除了。

沒有相關(guān)日志
存在異常，但最外層也捕獲了，所以不管有無異常都會 ACK。

第三個(gè)也在消費(fèi)的入口和提交消息出計(jì)算了時(shí)間，最終發(fā)現(xiàn)都是在2s左右 ACK 的。

偽代碼如下：

Consumer consumer = client.newConsumer()
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .topic(topic)
                .ackTimeout(30, TimeUnit.SECONDS)
                .subscriptionName("my-sub")
                .messageListener(new MessageListener<byte[]>() {
                    @SneakyThrows
                    @Override
                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                        log.info("msg_id{}",msg.getMessageId().toString());
                        TimeUnit.SECONDS.sleep(2);
                        consumer.acknowledge(msg);
                    }
                })
                .subscribe();

那這就很奇怪了，因?yàn)榇a里配置的 ackTimeout 是 30s，理論上來說是不會存在超時(shí)導(dǎo)致消息重發(fā)的。

為了排除是否是超時(shí)引起的，直接將業(yè)務(wù)代碼注釋掉了，等于是消息收到后立即就 ACK，經(jīng)過測試發(fā)現(xiàn)這樣確實(shí)就沒有重復(fù)消費(fèi)了。

為了再次確認(rèn)是不是和 ackTimeout 有關(guān)，直接將 .ackTimeout(30, TimeUnit.SECONDS) 注釋掉后測試，發(fā)現(xiàn)也沒有重復(fù)消費(fèi)了。

確認(rèn)原因

既然如此那一定是和這個(gè)配置有關(guān)了，但看代碼確實(shí)沒有超時(shí)，為了定位具體原因只有去看 client 的源碼了。

這里簡單梳理下消息的消費(fèi)的流程：

根據(jù) .receiverQueueSize(1000) 的配置，默認(rèn)情況下 broker 會直接給客戶端推送 1000 條消息。
客戶端將這 1000 條消息保存到內(nèi)部隊(duì)列中。
如果使用同步消費(fèi) receive() 時(shí)，本質(zhì)上就是去 take 這個(gè)內(nèi)部隊(duì)列。
如果是使用的是 messageListener 異步消費(fèi)并配置 ackTimeout，每當(dāng)從隊(duì)列里獲得一條消息后便會把這條消息加入 UnAckedMessageTracker 內(nèi)部的一個(gè)時(shí)間輪中，定時(shí)檢測頂部是否存在消息，如果存在則會觸發(fā)重新投遞。
4.1 加入時(shí)間輪后，異步調(diào)用我們自定義的事件，這個(gè)異步操作是提交到一個(gè)無界隊(duì)列中由單個(gè)線程依次排隊(duì)執(zhí)行（這點(diǎn)是這次問題的關(guān)鍵）
業(yè)務(wù) ACK 的時(shí)候會從時(shí)間輪中刪除消息，所以如果消息 ACK 的足夠快，在第四步就不會獲取到消息進(jìn)行重新投遞。

整體流程如上圖，代碼細(xì)節(jié)如下圖：

所以問題的根本原因就是寫入時(shí)間輪（UnAckedMessageTracker）開始倒計(jì)時(shí)的線程和回調(diào)業(yè)務(wù)邏輯的不是同一個(gè)線程。

如果業(yè)務(wù)執(zhí)行耗時(shí)，等到消息從那個(gè)單線程的無界隊(duì)列中取出來的時(shí)候很有可能已經(jīng)過了 ackTimeou 的時(shí)間，從而導(dǎo)致了超時(shí)重發(fā)。

也就是用戶所理解的 ackTimeout 周期（應(yīng)該進(jìn)入回調(diào)時(shí)候開始計(jì)時(shí)）和 SDK 實(shí)現(xiàn)的不一致造成的。

之后我再次確認(rèn)同樣的代碼換為同步消費(fèi)是沒有問題的，不會導(dǎo)致重復(fù)消費(fèi)：

while (true) {
Message msg = consumer.receive();
            log.info(
                    "consumer Message received: " + new String(msg.getData()) + msg.getMessageId().toString());
            TimeUnit.SECONDS.sleep(2);
            consumer.acknowledge(msg);    
}

查看代碼后發(fā)現(xiàn)同步代碼的獲取消息和加入 UnAckedMessageTracker 時(shí)間輪是同步的，也就不會出現(xiàn)超時(shí)的問題。