深入解析Java中的InterruptedException从异常处理到最佳实践方案

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. 消费者异步拉取消息
DefaultPullConsumerImpl.pullMessageAsync()
  → QueueSelector.selectQueueByTopic()
    → QueueSelector.refreshRoute()
      → RemotingApiWrapper.sync()
        → CommunicationAbstract.invokeSyncImpl()
          → ResponseFuture.getResponseUnit()
            → CountDownLatch.await()  // 在此处被中断

try {
    countDownLatch.await();
} catch (InterruptedException e) {
    // 恢复中断状态
    Thread.currentThread().interrupt();
    // 清理资源
    closeResources();
    // 可以选择重试或抛出业务异常
    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(); // 恢复中断状态
                running = false; // 准备退出
                log.warn("Consumer interrupted, shutting down...");
            } catch (Exception e) {
                log.error("Error processing message", e);
                sleepWithBackoff(); // 指数退避
            }
        }
    }
    private void sleepWithBackoff() {
        try {
            Thread.sleep(1000); // 简单示例，实际可用指数退避
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    public void stop() {
        running = false;
    }
}

ExecutorService executor = Executors.newFixedThreadPool(4);
// 提交任务
Future<?> future = executor.submit(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        try {
            // 执行任务
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断
            break;
        }
    }
});
// 关闭线程池
executor.shutdown();
try {
    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        executor.shutdownNow(); // 强制中断
    }
} catch (InterruptedException e) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

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