后端稳定性｜容错四大金刚：超时（五）MQ、异步、线程池阻塞全套治理（超时系列收官）

// ProducerImpl.java
public SendReceipt send(Message message) throws ClientException {
    final ListenableFuture<SendReceipt> future = Futures.transform(
        send(Collections.singletonList(message), false),
        sendReceipts -> sendReceipts.iterator().next(),
        MoreExecutors.directExecutor()
    );
    return handleClientFuture(future);  // 同步阻塞入口
}

// ClientImpl.java
protected <T> T handleClientFuture(ListenableFuture<T> future) throws ClientException {
    try {
        return future.get();  // 无业务层超时参数，会无限阻塞
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

// ClientConfigurationBuilder.java
private Duration requestTimeout = Duration.ofSeconds(3);  // 默认3秒
// 自定义配置示例
ClientConfiguration config = ClientConfiguration.newBuilder()
    .setEndpoints("xxx")
    .setRequestTimeout(Duration.ofSeconds(3))
    .build();

// KafkaProducer.doSend()
clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
RecordAccumulator.RecordAppendResult result = accumulator.append(...);

// BufferPool 内存分配精确递减计时
long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);
while (accumulated < size) {
    long startWaitNs = time.nanoseconds();
    boolean waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
    if (waitingTimeElapsed) {
        throw new BufferExhaustedException("缓冲区分配超时");
    }
    remainingTimeToBlockNs -= timeNs;
}

// NetworkClient.poll()
this.selector.poll(Utils.min(timeout, metadataTimeout, telemetryTimeout, defaultRequestTimeoutMs));

// 超时连接清理逻辑
private void handleTimedOutRequests(List<ClientResponse> responses, long now) {
    List<String> nodeIds = this.inFlightRequests.nodesWithTimedOutRequests(now);
    for (String nodeId : nodeIds) {
        this.selector.close(nodeId);
        log.info("节点请求超时，断开连接");
        processTimeoutDisconnection(responses, nodeId, now);
    }
}

public interface MessageListener {
    ConsumeResult consume(MessageView messageView);
}

@Component
public class OrderMessageListener {
    @Autowired
    private ThreadPoolExecutor orderConsumePool;

    @KafkaListener(topics = "order-topic", groupId = "order-consumer")
    public void onMessage(ConsumerRecord<String, String> record) {
        Future<?> future = orderConsumePool.submit(() -> processOrder(record.value()));
        try {
            future.get(800, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            log.error("订单消息处理超时，转入死信队列，record={}", record);
            future.cancel(true);
            sendToDLQ(record);
        } catch (Exception e) {
            log.error("消息处理异常", record, e);
            throw new RuntimeException(e);
        }
    }

    private void process(String msg) {
        // 订单更新、库存扣减业务逻辑
    }
}

Future<Result> future = executor.submit(task);
Result result = future.get();

Future<Result> future = executor.submit(task);
try {
    Result result = future.get(3, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    log.error("异步任务执行超时");
    future.cancel(true);
    result = Result.defaultValue();
}

private static final long MAX_SEND_WAIT_TIME_MS = 500;
ListenableFuture<SendResult<String, Object>> future = template.send(topic, key, data);
try {
    return future.get(MAX_SEND_WAIT_TIME_MS, TimeUnit.MILLISECONDS);
} catch (Exception e) {
    log.error("消息发送失败", topic, key, data, e);
}

CountDownLatch latch = new CountDownLatch(3);
latch.await();

CountDownLatch latch = new CountDownLatch(3);
boolean finished = latch.await(5, TimeUnit.SECONDS);
if (!finished) {
    log.error("子任务未全部完成");
    // 降级处理
}

Semaphore semaphore = new Semaphore(10);
semaphore.acquire();

boolean acquireOk = semaphore.tryAcquire(1, 1, TimeUnit.SECONDS);
if (!acquireOk) {
    throw new BizException("系统繁忙，请稍后重试");
}
try {
    // 业务逻辑
} finally {
    semaphore.release();
}

BlockingQueue<Task> queue = new LinkedBlockingQueue<>();
Task task = queue.take();

Task task = queue.poll(5, TimeUnit.SECONDS);
if (task == null) {
    log.warn("队列无任务，循环等待");
    continue;
}

workerThread.join();

workerThread.join(5000);
if (workerThread.isAlive()) {
    log.error("工作线程未正常退出");
    workerThread.interrupt();
}

final Timer closeTimer = time.timer(timeout);
this.sender.initiateClose();
closeTimer.update();
if (this.ioThread != null) {
    this.ioThread.join(closeTimer.remainingMs());
}

@Async("taskExecutor")
public Future<String> asyncTask() {
    return new AsyncResult<>("done");
}
// 无超时阻塞
Future<String> future = asyncTask();
future.get();

@Async("taskExecutor")
public Future<String> asyncTask() {
    return new AsyncResult<>("done");
}
// 调用层强制限时
Future<String> future = asyncTask();
try {
    String res = future.get(3, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    log.error("异步任务超时");
    future.cancel(true);
}

spring:
  task:
    execution:
      shutdown:
        await-termination: true
        await-termination-period: 10s

@XxlJob("dailyReportJob")
public ReturnT<String> execute() {
    String lockKey = "job:dailyReport";
    boolean locked = redisLock.tryLock(lockKey, 30, TimeUnit.SECONDS);
    if (!locked) {
        log.warn("未获取定时任务锁，跳过本次执行");
        return ReturnT.SUCCESS;
    }
    ExecutorService pool = Executors.newFixedThreadPool(4);
    Future<?> future = pool.submit(this::generateDailyReport);
    try {
        future.get(25, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        log.error("定时任务执行超时");
        future.cancel(true);
        return ReturnT.FAIL;
    } catch (Exception e) {
        log.error("定时任务异常", e);
        return ReturnT.FAIL;
    } finally {
        redisLock.unlock(lockKey);
        pool.shutdown();
        if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }
    }
    return ReturnT.SUCCESS;
}

@Scheduled(fixedRate = 30000)
public void syncDataTask() {
    String lockKey = "schedule:sync";
    boolean locked = redisLock.tryLock(lockKey, 30, TimeUnit.SECONDS);
    if (!locked) return;
    try {
        CompletableFuture<Void> task = CompletableFuture.run(this.syncBusiness, taskExecutor);
        task.get(25, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        log.error("定时同步任务执行超时");
    } finally {
        redisLock.unlock(lockKey);
    }
}