通過 Pulsar 源碼徹底解決重復消費問題

作者：crossoverJie 2023-02-27 08:08:54

開發前端

如果業務執行耗時，等到消息從那個單線程的無界隊列中取出來的時候很有可能已經過了 ackTimeou 的時間，從而導致了超時重發。

背景

最近真是和 Pulsar 杠上了，業務團隊反饋說是線上有個應用消息重復消費。

而且在測試環境是可以穩定復現的，根據經驗來看一般能穩定復現的都比較好解決。

定位問題

接著便是定位問題了，根據之前的經驗讓業務按照這幾種情況先排查一下：

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

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

第三個也在消費的入口和提交消息出計算了時間，最終發現都是在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，理論上來說是不會存在超時導致消息重發的。

為了排除是否是超時引起的，直接將業務代碼注釋掉了，等于是消息收到后立即就 ACK，經過測試發現這樣確實就沒有重復消費了。

為了再次確認是不是和 ackTimeout 有關，直接將 .ackTimeout(30, TimeUnit.SECONDS) 注釋掉后測試，發現也沒有重復消費了。

確認原因

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

這里簡單梳理下消息的消費的流程：

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

整體流程如上圖，代碼細節如下圖：

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

如果業務執行耗時，等到消息從那個單線程的無界隊列中取出來的時候很有可能已經過了 ackTimeou 的時間，從而導致了超時重發。

也就是用戶所理解的 ackTimeout 周期（應該進入回調時候開始計時）和 SDK 實現的不一致造成的。

之后我再次確認同樣的代碼換為同步消費是沒有問題的，不會導致重復消費：

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); 
}