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

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

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

背景

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

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

定位問題

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

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

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

第三個也在消費(fèi)的入口和提交消息出計算了時間，最終發(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();

那這就很奇怪了，因為代碼里配置的 ackTimeout 是 30s，理論上來說是不會存在超時導(dǎo)致消息重發(fā)的。

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

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

確認(rèn)原因

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

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

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

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

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

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

也就是用戶所理解的 ackTimeout 周期（應(yīng)該進(jìn)入回調(diào)時候開始計時）和 SDK 實現(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 時間輪是同步的，也就不會出現(xiàn)超時的問題。