java中Pulsar?InterruptedException?異常

更新時間：2023年02月23日 11:50:32 作者：crossoverJie

這篇文章主要為大家介紹了java中Pulsar?InterruptedException?異常分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

背景

今天收到業(yè)務團隊反饋線上有個應用往 Pulsar 中發(fā)送消息失敗了，經(jīng)過日志查看得知是發(fā)送消息時候拋出了 java.lang.InterruptedException 異常。

和業(yè)務溝通后得知是在一個 gRPC 接口中觸發(fā)的消息發(fā)送，大約持續(xù)了半個小時的異常后便恢復正常了，這是整個問題的背景。

前置排查

拿到該問題后首先排查下是否是共性問題，查看了其他的應用沒有發(fā)現(xiàn)類似的異常；同時也查看了 Pulsar broker 的監(jiān)控大盤，在這個時間段依然沒有波動和異常；

這樣可以初步排除是 Pulsar 服務端的問題。

接著便是查看應用那段時間的負載情況，從應用 QPS 到 JVM 的各個內(nèi)存情況依然沒發(fā)現(xiàn)有什么明顯的變化。

Pulsar 源碼排查

既然看起來應用本身和 Pulsar broker 都沒有問題的話那就只能從異常本身來排查了。

首先第一步要得知具體使用的是 Pulsar-client 是版本是多少，因為業(yè)務使用的是內(nèi)部基于官方 SDK 封裝 springboot starter 所以第一步還得排查這個 starter 是否有影響。

通過查看源碼基本排除了 starter 的嫌疑，里面只是簡單的封裝了 SDK 的功能而已。

org.apache.pulsar.client.api.PulsarClientException: java.util.concurrent.ExecutionException: org.apache.pulsar.client.api.PulsarClientException: java.lang.InterruptedException at org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:1027) at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.send(TypedMessageBuilderImpl.java:91) at 
java.base/java.lang.Thread.run(Thread.java:834) Caused by: java.util.concurrent.ExecutionException: org.apache.pulsar.client.api.PulsarClientException: java.lang.InterruptedException at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) 
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) 
at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.send(TypedMessageBuilderImpl.java:89) ... 49 common frames omitted Caused by: org.apache.pulsar.client.api.PulsarClientException: java.lang.InterruptedException 
at org.apache.pulsar.client.impl.ProducerImpl.canEnqueueRequest(ProducerImpl.java:775) 
at org.apache.pulsar.client.impl.ProducerImpl.sendAsync$original$BWm7PPlZ(ProducerImpl.java:393) 
at org.apache.pulsar.client.impl.ProducerImpl.sendAsync$original$BWm7PPlZ$accessor$i7NYMN6i(ProducerImpl.java) 
at org.apache.pulsar.client.impl.ProducerImpl$auxiliary$EfuVvJLT.call(Unknown Source) 
at org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstMethodsInter.intercept(InstMethodsInter.java:86) 
at org.apache.pulsar.client.impl.ProducerImpl.sendAsync(ProducerImpl.java) 
at org.apache.pulsar.client.impl.ProducerImpl.internalSendAsync(ProducerImpl.java:292) 
at org.apache.pulsar.client.impl.ProducerImpl.internalSendWithTxnAsync(ProducerImpl.java:363) 
at org.apache.pulsar.client.impl.PartitionedProducerImpl.internalSendWithTxnAsync(PartitionedProducerImpl.java:191) 
at org.apache.pulsar.client.impl.PartitionedProducerImpl.internalSendAsync(PartitionedProducerImpl.java:167) 
at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.sendAsync(TypedMessageBuilderImpl.java:103) 
at org.apache.pulsar.client.impl.TypedMessageBuilderImpl.send(TypedMessageBuilderImpl.java:82) ... 49 common frames omitted Caused by: java.lang.InterruptedException: null
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1343) 
at java.base/java.util.concurrent.Semaphore.acquire(Semaphore.java:318) 
at org.apache.pulsar.client.impl.ProducerImpl.canEnqueueRequest(ProducerImpl.java:758)

接下來便只能是分析堆棧了，因為 Pulsar-client 的部分實現(xiàn)源碼是沒有直接打包到依賴中的，反編譯的話許多代碼行數(shù)對不上，所以需要將官方的源碼拉到本地，切換到對于的分支進行查看。

這一步稍微有點麻煩，首先是代碼庫還挺大的，加上之前如果沒有準備好 Pulsar 的開發(fā)環(huán)境的話估計會勸退一部分人；

但其實大部分問題都是網(wǎng)絡造成的，只要配置一些 Maven 鏡像多試幾次總會編譯成功。

我這里直接將分支切換到 branch-2.8。

從堆棧的頂部開始排查 TypedMessageBuilderImpl.java:91：

看起來是內(nèi)部異步發(fā)送消息的時候拋了異常。

接著往下看到這里：

java.lang.InterruptedException 
at org.apache.pulsar.client.impl.ProducerImpl.canEnqueueRequest(ProducerImpl.java:775) at

看起來是這里沒錯，但是代碼行數(shù)明顯不對；因為 2.8 這個分支也是修復過幾個版本，所以中間有修改導致代碼行數(shù)與最新代碼對不上也正常。

semaphore.get().acquire();

不過初步來看應該是這行代碼拋出的線程終端異常，這里看起來只有他最有可能了。

為了確認是否是真的是這行代碼，這個文件再往前翻了幾個版本最終確認了就是這行代碼沒錯了。

我們點開java.util.concurrent.Semaphore#acquire()的源碼，

    /**
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@linkplain Thread#interrupt interrupted} while waiting
     * for a permit,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     *
     * @throws InterruptedException if the current thread is interrupted
     */
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
        if (Thread.interrupted() ||
            (tryAcquireShared(arg) < 0 &&
             acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }

通過源碼會發(fā)現(xiàn) acquire() 函數(shù)確實會響應中斷，一旦檢測到當前線程被中斷后便會拋出 InterruptedException 異常。

定位問題

所以問題的原因基本確定了，就是在 Pulsar 的發(fā)送消息線程被中斷了導致的，但為啥會被中斷還需要繼續(xù)排查。

我們知道線程中斷是需要調(diào)用 Thread.currentThread().interrupt(); API的，首先猜測是否 Pulsar 客戶端內(nèi)部有個線程中斷了這個發(fā)送線程。

于是我在 pulsar-client 這個模塊中搜索了相關代碼：

排除掉和 producer 不相關的地方，其余所有中斷線程的代碼都是在有了該異常之后繼續(xù)傳遞而已；所以初步來看 pulsar-client 內(nèi)部沒有主動中斷的操作。

既然 Pulsar 自己沒有做，那就只可能是業(yè)務做的了？

于是我在業(yè)務代碼中搜索了一下：

果然在業(yè)務代碼中搜到了唯一一處中斷的地方，而且通過調(diào)用關系得知這段代碼是在消息發(fā)送前執(zhí)行的，并且和 Pulsar 發(fā)送函數(shù)處于同一線程。

大概的偽代碼如下：

        List.of(1, 2, 3).stream().map(e -> {
                    return CompletableFuture.supplyAsync(() -> {
                        try {
                            TimeUnit.MILLISECONDS.sleep(10);
                        } catch (InterruptedException ex) {
                            throw new RuntimeException(ex);
                        }
                        return e;
                    });
                }
        ).collect(Collectors.toList()).forEach(f -> {
            try {
                Integer integer = f.get();
                log.info("====" + integer);
                if (integer==3){
                    TimeUnit.SECONDS.sleep(10);
                    Thread.currentThread().interrupt();
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        });
	   MessageId send = producer.newMessage().value(msg.getBytes()).send();

執(zhí)行這段代碼可以完全復現(xiàn)同樣的堆棧。

幸好中斷這里還打得有日志：

通過日志搜索發(fā)現(xiàn)異常的時間和這個中斷的日志時間點完全重合，這樣也就知道根本原因了。

因為業(yè)務線程和消息發(fā)送線程是同一個，在某些情況下會執(zhí)行 Thread.currentThread().interrupt();，其實單純執(zhí)行這行函數(shù)并不會發(fā)生什么，只要沒有去響應這個中斷，也就是 Semaphore 源碼中的判斷了線程中斷的標記：

    public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
        if (Thread.interrupted() ||
            (tryAcquireShared(arg) < 0 &&
             acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }

但恰好這里業(yè)務中斷后自己并沒有去判斷這個標記，導致 Pulsar 內(nèi)部去判斷了，最終拋出了這個異常。