线程池深度解析：工作机制、异常处理与内存泄漏解决方案

ExecutorService fixedPool = Executors.newFixedThreadPool(5);

ExecutorService cachedPool = Executors.newCachedThreadPool();

ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);

ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();

executor.submit(() -> {
    try {
        // 作战代码
    } catch (CombatException e) {
        // 现场急救处理
        log.error("前线异常：", e);
        // 发送异常报告
        alertSystem.notify(e);
    }
});

ThreadFactory factory = r -> {
    Thread t = new Thread(r);
    t.setUncaughtExceptionHandler((thread, ex) -> {
        log.error("线程{}发生未捕获异常：", thread.getName(), ex);
        // 自动重启线程
        executor.execute(thread.getRunnable());
    });
    return t;
};
ExecutorService pool = new ThreadPoolExecutor(..., factory);

Future<?> future = executor.submit(task);
try {
    future.get();
} catch (ExecutionException e) {
    Throwable rootCause = e.getCause();
    log.error("任务执行异常：", rootCause);
    // 启动应急方案
    contingencyPlan.execute();
}

public class RetryTask implements Runnable {
    private static final int MAX_RETRY = 3;
    private int retryCount = 0;
    
    @Override
    public void run() {
        try {
            processOrder();
        } catch (TransientException e) {
            if (retryCount++ < MAX_RETRY) {
                long delay = (long) Math.pow(2, retryCount);
                executor.schedule(this, delay, TimeUnit.SECONDS);
            } else {
                alertService.notifyCritical(e);
            }
        }
    }
    
    private void processOrder() throws TransientException {
        // 订单处理逻辑
    }
}

executor.execute(() -> {
    Connection conn = dataSource.getConnection(); // 获取连接
    try {
        // 执行数据库操作
    } catch (SQLException e) {
        // 处理异常
    }
    // 忘记conn.close()
});

  try (Connection conn = dataSource.getConnection();
       Statement stmt = conn.createStatement()) {
      // 执行操作
  }

private static Map<Long, TaskContext> CACHE = new ConcurrentHashMap<>();

executor.submit(() -> {
    TaskContext context = new TaskContext();
    CACHE.put(Thread.currentThread().getId(), context);
    // 执行任务...
    // 忘记移除缓存项
});

private ThreadLocal<BigObject> threadLocal = new ThreadLocal<>();

executor.execute(() -> {
    threadLocal.set(new BigObject()); // 大对象
    // 使用后未remove
});

  try {
      threadLocal.set(new BigObject());
      // 业务逻辑
  } finally {
      threadLocal.remove();
  }

   jmap -dump:format=b,file=heap.bin <pid>

public void safeExecute(Runnable task) {
    executor.submit(() -> {
        try {
            // 前置资源申请
            Resource resource = acquireResource();
            try {
                task.run();
            } finally {
                // 确保资源释放
                resource.close();
            }
        } catch (Exception e) {
            exceptionHandler.handle(e);
        }
    });
}

// 注册内存监控Bean
@Bean
public MemoryMonitor memoryMonitor() {
    return new MemoryMonitor(executor) {
        @Override
        protected void onMemoryThresholdExceeded() {
            // 自动扩容或报警
            scalingService.scaleOut();
            alertService.sendCriticalAlert();
        }
    };
}

@Bean("orderProcessor")
public ThreadPoolTaskExecutor orderExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(20);      // 常规处理能力
    executor.setMaxPoolSize(100);      // 大促期间扩容能力
    executor.setQueueCapacity(500);    // 应对突发流量
    executor.setKeepAliveSeconds(60);  // 空闲线程回收时间
    executor.setThreadNamePrefix("Order-Processor-");
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.initialize();
    return executor;
}

@Slf4j
public class OrderExceptionHandler implements AsyncUncaughtExceptionHandler {
    private final RetryTemplate retryTemplate;
    private final DeadLetterQueue deadLetterQueue;

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        log.error("订单处理异常 [方法:{}]: {}", method.getName(), ex.getMessage());
        
        if (isRecoverable(ex)) {
            retryTemplate.execute(context -> {
                // 重试逻辑
                return processWithRetry(params);
            });
        } else {
            deadLetterQueue.put(new DeadLetter(method, params, ex));
            alertService.notifyTeam(ex);
        }
    }

    private boolean isRecoverable(Throwable ex) {
        return ex instanceof TransientException 
            || ex.getCause() instanceof TimeoutException;
    }
}

   if (MemoryMonitor.getUsedHeap() > 80%) {
       flowControl.throttleRequests();
   }

   @Scheduled(fixedRate = 5 * 60 * 1000)
   public void performHealthCheck() {
       checkThreadPoolStatus();
       analyzeMemoryUsage();
       verifyResourceLeaks();
   }

public class DynamicThreadPool implements Runnable {
    private ThreadPoolExecutor executor;
    private ScheduledExecutorService adjuster;

    public void startAdjusting() {
        adjuster.scheduleAtFixedRate(this::adjustPool, 30, 30, TimeUnit.SECONDS);
    }

    private void adjustPool() {
        int currentQueueSize = executor.getQueue().size();
        int currentActive = executor.getActiveCount();
        
        if (currentQueueSize > queueHighWaterMark) {
            int delta = (currentQueueSize - queueHighWaterMark) / tasksPerThread;
            int newCoreSize = Math.min(corePoolSize + delta, maxPoolSize);
            executor.setCorePoolSize(newCoreSize);
        } else if (currentActive < corePoolSize * 0.7) {
            executor.setCorePoolSize(Math.max(minCoreSize, (int)(corePoolSize * 0.9)));
        }
    }
}

   public class ContextAwareExecutor implements Executor {
       private final Executor delegate;
       
       public void execute(Runnable command) {
           Map<String, Object> context = ContextHolder.getCurrentContext();
           delegate.execute(() -> {
               ContextHolder.setContext(context);
               try {
                   command.run();
               } finally {
                   ContextHolder.clear();
               }
           });
       }
   }