一、问题背景：线程池耗尽的"死亡螺旋"

你是否遇到过这样的场景：系统使用 @Async 异步执行任务，在流量高峰期突然出现大量请求超时，最终导致整个服务不可用？

这很可能是 Future.get() 阻塞导致的线程池雪崩。当异步任务的处理速度跟不上请求速度时，任务会在队列中堆积，而调用线程在 Future.get() 处阻塞等待，最终导致调用线程也被耗尽。

// 危险的异步调用方式
@Async
public CompletableFuture<String> doAsyncTask() {
    // 执行耗时操作...
}

// 调用方
public void process() {
    CompletableFuture<String> future = asyncService.doAsyncTask();
    String result = future.get(); // 阻塞等待，可能导致线程耗尽
}

真实案例：某支付系统在双十一期间，异步对账任务因数据库慢查询导致处理延迟，调用线程在 Future.get() 处阻塞，最终导致 Tomcat 线程池被占满，新请求无法处理。

二、核心概念：线程池雪崩原理

2.1 雪崩形成过程

┌──────────────────────────────────────────────────────────────────┐
│                    请求处理流程                                   │
├──────────────────────────────────────────────────────────────────┤
│                                                                 │
│  请求 ──► Tomcat线程 ──► Future.get() ──┐                       │
│                                       │                         │
│                    ┌───────────────────▼───────────────────┐    │
│                    │        异步线程池（有限容量）            │    │
│                    │  ┌─────────────────────────────────┐  │    │
│                    │  │  Task 1   Task 2   Task 3 ...  │  │    │
│                    │  └─────────────────────────────────┘  │    │
│                    └───────────────────────────────────────┘    │
│                              │                                  │
│                              ▼                                  │
│                    任务队列（可能无限增长）                        │
│                              │                                  │
│                              ▼                                  │
│                    系统资源耗尽 → 服务雪崩                        │
└──────────────────────────────────────────────────────────────────┘

2.2 关键问题分析

三、实现方案：队列限制 + 降级回调

3.1 方案架构设计

┌──────────────────────────────────────────────────────────────────┐
│                    异步任务处理架构                                │
├──────────────────────────────────────────────────────────────────┤
│                                                                 │
│  请求 ──► 限流层 ──► 队列满？ ──┬── 是 ──► 降级回调              │
│                                  │                              │
│                                  ▼                              │
│                         异步线程池                                │
│                         (有界队列 + 拒绝策略)                      │
│                                  │                              │
│                                  ▼                              │
│                         Future.get(timeout)                      │
│                                  │                              │
│                                  ▼                              │
│                         结果处理/超时处理                         │
└──────────────────────────────────────────────────────────────────┘

3.2 线程池配置优化

创建自定义异步线程池配置：

@Configuration
@EnableAsync
public class AsyncPoolConfig {
    
    private static final Logger log = LoggerFactory.getLogger(AsyncPoolConfig.class);
    
    /**
     * 核心线程数：CPU核心数
     */
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
    
    /**
     * 最大线程数：CPU核心数 × 2
     */
    private static final int MAX_POOL_SIZE = CORE_POOL_SIZE * 2;
    
    /**
     * 队列容量：100（有界队列，避免内存溢出）
     */
    private static final int QUEUE_CAPACITY = 100;
    
    /**
     * 空闲线程存活时间：60秒
     */
    private static final int KEEP_ALIVE_SECONDS = 60;
    
    @Bean("customAsyncExecutor")
    public Executor customAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        
        // 核心线程数
        executor.setCorePoolSize(CORE_POOL_SIZE);
        
        // 最大线程数
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        
        // 有界队列
        executor.setQueueCapacity(QUEUE_CAPACITY);
        
        // 空闲线程存活时间
        executor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        
        // 线程名称前缀
        executor.setThreadNamePrefix("async-executor-");
        
        // 拒绝策略：调用者运行（降级处理）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        
        // 等待所有任务完成后再关闭
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        
        // 任务执行前后的回调
        executor.setTaskDecorator(runnable -> {
            String taskName = Thread.currentThread().getName();
            log.info("Start executing task: {}", taskName);
            long startTime = System.currentTimeMillis();
            
            return () -> {
                try {
                    runnable.run();
                } finally {
                    long duration = System.currentTimeMillis() - startTime;
                    log.info("Task {} completed in {}ms", taskName, duration);
                }
            };
        });
        
        executor.initialize();
        return executor;
    }
}

3.3 降级回调机制实现

创建异步任务包装器，支持超时控制和降级处理：

@Component
public class AsyncTaskWrapper {
    
    private static final Logger log = LoggerFactory.getLogger(AsyncTaskWrapper.class);
    
    /**
     * 执行异步任务，带超时控制和降级回调
     * 
     * @param taskSupplier 任务提供者
     * @param timeout 超时时间
     * @param fallback 降级回调
     * @param <T> 返回值类型
     * @return 任务结果或降级结果
     */
    public <T> T executeWithFallback(Supplier<T> taskSupplier, 
                                      Duration timeout,
                                      Supplier<T> fallback) {
        CompletableFuture<T> future = CompletableFuture.supplyAsync(taskSupplier);
        
        try {
            return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            log.warn("Async task timeout after {}ms, invoking fallback", timeout.toMillis());
            future.cancel(true);
            return fallback.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("Async task interrupted, invoking fallback");
            return fallback.get();
        } catch (ExecutionException e) {
            log.error("Async task execution failed, invoking fallback", e);
            return fallback.get();
        }
    }
    
    /**
     * 执行异步任务，带超时控制
     */
    public <T> CompletableFuture<T> executeWithTimeout(Supplier<T> taskSupplier, 
                                                        Duration timeout) {
        return CompletableFuture.supplyAsync(taskSupplier)
            .completeOnTimeout(null, timeout.toMillis(), TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                log.error("Async task failed", ex);
                return null;
            });
    }
}

3.4 异步任务定义

@Service
public class AsyncService {
    
    private static final Logger log = LoggerFactory.getLogger(AsyncService.class);
    
    @Async("customAsyncExecutor")
    public CompletableFuture<String> processOrder(String orderId) {
        log.info("Processing order: {}", orderId);
        
        // 模拟耗时操作
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("Order processing interrupted: {}", orderId);
            return CompletableFuture.completedFuture("Interrupted");
        }
        
        log.info("Order processed: {}", orderId);
        return CompletableFuture.completedFuture("Success: " + orderId);
    }
    
    @Async("customAsyncExecutor")
    public CompletableFuture<String> fetchUserInfo(String userId) {
        log.info("Fetching user info: {}", userId);
        
        // 模拟可能超时的操作
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return CompletableFuture.completedFuture("Interrupted");
        }
        
        return CompletableFuture.completedFuture("User: " + userId);
    }
}

3.5 调用方使用示例

@Service
public class OrderService {
    
    @Autowired
    private AsyncService asyncService;
    
    @Autowired
    private AsyncTaskWrapper asyncTaskWrapper;
    
    /**
     * 处理订单，带降级策略
     */
    public String handleOrder(String orderId) {
        return asyncTaskWrapper.executeWithFallback(
            () -> {
                CompletableFuture<String> future = asyncService.processOrder(orderId);
                return future.get(); // 内部已处理异常
            },
            Duration.ofSeconds(5),
            () -> {
                log.warn("Order processing fallback triggered for: {}", orderId);
                return "Fallback: Order will be processed later";
            }
        );
    }
    
    /**
     * 批量处理订单，使用 CompletableFuture.allOf
     */
    public List<String> batchHandleOrders(List<String> orderIds) {
        List<CompletableFuture<String>> futures = orderIds.stream()
            .map(orderId -> asyncService.processOrder(orderId))
            .collect(Collectors.toList());
        
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        
        try {
            // 等待所有任务完成，带超时
            allFutures.get(30, TimeUnit.SECONDS);
            
            return futures.stream()
                .map(future -> {
                    try {
                        return future.get();
                    } catch (Exception e) {
                        return "Failed";
                    }
                })
                .collect(Collectors.toList());
        } catch (TimeoutException e) {
            log.warn("Batch order processing timeout");
            return Collections.singletonList("Batch timeout");
        } catch (Exception e) {
            log.error("Batch order processing failed", e);
            return Collections.singletonList("Batch failed");
        }
    }
}

四、监控与告警

4.1 线程池监控指标

创建监控组件：

@Component
public class AsyncPoolMetrics {
    
    private final MeterRegistry meterRegistry;
    private final ThreadPoolTaskExecutor executor;
    
    public AsyncPoolMetrics(MeterRegistry meterRegistry, 
                           @Qualifier("customAsyncExecutor") ThreadPoolTaskExecutor executor) {
        this.meterRegistry = meterRegistry;
        this.executor = executor;
        
        // 注册监控指标
        registerMetrics();
    }
    
    private void registerMetrics() {
        // 活跃线程数
        Gauge.builder("async.pool.active_threads", 
            () -> executor.getActiveCount())
            .register(meterRegistry);
        
        // 队列大小
        Gauge.builder("async.pool.queue_size", 
            () -> executor.getThreadPoolExecutor().getQueue().size())
            .register(meterRegistry);
        
        // 队列剩余容量
        Gauge.builder("async.pool.queue_remaining", 
            () -> executor.getThreadPoolExecutor().getQueue().remainingCapacity())
            .register(meterRegistry);
        
        // 已完成任务数
        Gauge.builder("async.pool.completed_tasks", 
            () -> executor.getThreadPoolExecutor().getCompletedTaskCount())
            .register(meterRegistry);
        
        // 线程池大小
        Gauge.builder("async.pool.pool_size", 
            () -> executor.getPoolSize())
            .register(meterRegistry);
    }
}

4.2 Prometheus 告警规则

groups:
- name: async_pool_alerts
  rules:
  - alert: AsyncPoolQueueFull
    expr: async_pool_queue_remaining <= 10
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "异步线程池队列即将满"
      description: "队列剩余容量: {{ $value }}"
  
  - alert: AsyncPoolHighActiveThreads
    expr: async_pool_active_threads / async_pool_pool_size > 0.8
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "异步线程池活跃线程占比过高"
      description: "活跃线程数: {{ $value }}"
  
  - alert: AsyncPoolRejectedTasks
    expr: rate(async_pool_rejected_total[5m]) > 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "异步任务被拒绝"
      description: "最近5分钟有任务被拒绝"

五、配置文件示例

server:
  port: 8080

spring:
  application:
    name: async-pool-demo

# 异步线程池配置
async:
  pool:
    core-size: ${CPU_COUNT:4}
    max-size: ${CPU_COUNT:8}
    queue-capacity: 100
    keep-alive-seconds: 60

# 监控配置
management:
  endpoints:
    web:
      exposure:
        include: health, info, prometheus, metrics
  metrics:
    tags:
      application: ${spring.application.name}

logging:
  level:
    com.example.async: DEBUG

六、最佳实践建议

6.1 线程池参数调优指南

6.2 避免线程耗尽的关键点

1. 使用有界队列：永远不要使用无界队列
2. 设置超时时间：Future.get(timeout) 而非 Future.get()
3. 实现降级策略：队列满时有兜底方案
4. 监控线程池状态：及时发现异常
5. 避免嵌套异步：异步任务内不要再调用异步方法

6.3 常见陷阱

互动话题

您在生产环境中遇到过异步线程池耗尽的问题吗？您是如何解决的？欢迎在评论区分享您的经验！更多技术文章，欢迎关注公众号：服务端技术精选。