2025-06-20 01:08:37 [Thread-2] WARN  JcqCommunication - Exception occurs when sending one sync request to the remote address jcq-hb-yd-001-manager-nlb-FI.jvessel-open-hb.jdcloud.com:2888, but got exception java.lang.InterruptedException
2025-06-20 01:08:37 [Thread-2] WARN  c.j.j.c.common.RemotingApiWrapper - get exception when sync request to address:jcq-hb-yd-001-manager-nlb-FI.jvessel-open-hb.jdcloud.com:2888, request:GetTopicRouteInfoRequestV2{...}
2025-06-20 01:08:37 [Thread-2] WARN  c.j.j.c.common.RemotingApiWrapper - exception:
java.lang.InterruptedException: null
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1326)
	at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:277)
	at com.jcloud.jcq.communication.core.ResponseFuture.getResponseUnit(ResponseFuture.java:66)
	...

// 1. 消費(fèi)者異步拉取消息
DefaultPullConsumerImpl.pullMessageAsync()
  → QueueSelector.selectQueueByTopic()
    → QueueSelector.refreshRoute()
      → RemotingApiWrapper.sync()
        → CommunicationAbstract.invokeSyncImpl()
          → ResponseFuture.getResponseUnit()
            → CountDownLatch.await()  // 在此處被中斷

try {
    countDownLatch.await();
} catch (InterruptedException e) {
    // 恢復(fù)中斷狀態(tài)
    Thread.currentThread().interrupt();
    // 清理資源
    closeResources();
    // 可以選擇重試或拋出業(yè)務(wù)異常
    throw new BusinessException("Task interrupted", e);
}

public class RobustMQConsumer {
    private volatile boolean running = true;

    public void start() {
        while (running && !Thread.currentThread().isInterrupted()) {
            try {
                Message message = pullMessage();
                processMessage(message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢復(fù)中斷狀態(tài)
                running = false; // 準(zhǔn)備退出
                log.warn("Consumer interrupted, shutting down...");
            } catch (Exception e) {
                log.error("Error processing message", e);
                sleepWithBackoff(); // 指數(shù)退避
            }
        }
    }

    private void sleepWithBackoff() {
        try {
            Thread.sleep(1000); // 簡(jiǎn)單示例，實(shí)際可用指數(shù)退避
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public void stop() {
        running = false;
    }
}

ExecutorService executor = Executors.newFixedThreadPool(4);

// 提交任務(wù)
Future<?> future = executor.submit(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        try {
            // 執(zhí)行任務(wù)
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢復(fù)中斷
            break;
        }
    }
});

// 關(guān)閉線程池
executor.shutdown();
try {
    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        executor.shutdownNow(); // 強(qiáng)制中斷
    }
} catch (InterruptedException e) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    consumer.stop();
    executor.shutdown();
}));