背景

最近真是和 Pulsar 杠上了，業(yè)務(wù)團(tuán)隊反饋說是線上有個應(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。

第三個也在消費(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();

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

為了排除是否是超時引起的，直接將業(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)原因

既然如此那一定是和這個配置有關(guān)了，但看代碼確實(shí)沒有超時，為了定位具體原因只有去看 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í)行（這點(diǎn)是這次問題的關(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 實(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 時間輪是同步的，也就不會出現(xiàn)超時的問題。