线程池监控调优与排查案例

public ThreadPoolExecutor(
    int corePoolSize,              // 核心线程数
    int maximumPoolSize,           // 最大线程数
    long keepAliveTime,            // 空闲线程存活时间
    TimeUnit unit,                 // 时间单位
    BlockingQueue<Runnable> workQueue,  // 工作队列
    ThreadFactory threadFactory,   // 线程工厂
    RejectedExecutionHandler handler    // 拒绝策略
)

public class ThreadPoolMonitor {
    private ThreadPoolExecutor executor;
    
    public void printThreadPoolStats() {
        System.out.println("核心线程数: " + executor.getCorePoolSize());
        System.out.println("最大线程数: " + executor.getMaximumPoolSize());
        System.out.println("当前线程数: " + executor.getPoolSize());
        System.out.println("活跃线程数: " + executor.getActiveCount());
        System.out.println("历史最大线程数: " + executor.getLargestPoolSize());
        System.out.println("已完成任务数: " + executor.getCompletedTaskCount());
        System.out.println("总任务数: " + executor.getTaskCount());
        System.out.println("队列大小: " + executor.getQueue().size());
        System.out.println("队列剩余容量: " + executor.getQueue().remainingCapacity());
        
        // 计算线程池使用率
        double usage = (double) executor.getActiveCount() / executor.getMaximumPoolSize() * 100;
        System.out.println("线程池使用率: " + String.format("%.2f", usage) + "%");
        
        // 计算队列使用率
        double queueUsage = (double) executor.getQueue().size() / 
            (executor.getQueue().size() + executor.getQueue().remainingCapacity()) * 100;
        System.out.println("队列使用率: " + String.format("%.2f", queueUsage) + "%");
    }
}

public class MonitoredThreadPoolExecutor extends ThreadPoolExecutor {
    private final AtomicLong totalTaskTime = new AtomicLong();
    private final AtomicLong totalTaskCount = new AtomicLong();
    
    public MonitoredThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        System.out.println("任务开始执行: " + t.getName() + ", 时间: " + new Date());
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        System.out.println("任务执行完成, 时间: " + new Date());
        if (t != null) {
            System.out.println("任务执行异常: " + t.getMessage());
            t.printStackTrace();
        }
    }
    
    @Override
    protected void terminated() {
        super.terminated();
        System.out.println("线程池已终止");
    }
    
    public long getTotalTaskTime() {
        return totalTaskTime.get();
    }
    
    public long getTotalTaskCount() {
        return totalTaskCount.get();
    }
    
    public double getAverageTaskTime() {
        long count = totalTaskCount.get();
        return count == 0 ? 0 : (double) totalTaskTime.get() / count;
    }
}

public class ThreadPoolMetrics {
    private final ThreadPoolExecutor executor;
    private final String poolName;
    
    public ThreadPoolMetrics(ThreadPoolExecutor executor, String poolName) {
        this.executor = executor;
        this.poolName = poolName;
        startMonitor();
    }
    
    private void startMonitor() {
        ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
        monitor.scheduleAtFixedRate(() -> {
            // 上报到监控系统（如 Prometheus、Micrometer）
            Metrics.gauge("thread.pool.active", executor.getActiveCount(), "pool", poolName);
            Metrics.gauge("thread.pool.size", executor.getPoolSize(), "pool", poolName);
            Metrics.gauge("thread.pool.queue.size", executor.getQueue().size(), "pool", poolName);
            Metrics.gauge("thread.pool.completed", executor.getCompletedTaskCount(), "pool", poolName);
            
            // 告警逻辑
            if (executor.getQueue().size() > 100) {
                AlertService.sendAlert("线程池[" + poolName + "]队列积压: " + executor.getQueue().size());
            }
            
            double usage = (double) executor.getActiveCount() / executor.getMaximumPoolSize();
            if (usage > 0.8) {
                AlertService.sendAlert("线程池[" + poolName + "]使用率过高: " + usage);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }
}

// CPU 密集型任务
int cpuCount = Runtime.getRuntime().availableProcessors();
int corePoolSize = cpuCount + 1;

// IO 密集型任务
int corePoolSize = cpuCount * 2;

// 混合型任务
// 需要根据 CPU 时间和 IO 时间的比例调整
// 公式: 线程数 = CPU 核心数 * (1 + 等待时间 / 计算时间)
int corePoolSize = (int) (cpuCount * (1 + waitTime / computeTime));

// 有界队列: 防止 OOM,但可能触发拒绝策略
ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);

// 无界队列: 不会触发拒绝策略,但可能导致 OOM
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();

// 同步队列: 直接提交,不缓冲
SynchronousQueue<Runnable> queue = new SynchronousQueue<>();

// 优先级队列: 按优先级执行
PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>();

// 自定义拒绝策略
public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 记录日志
        System.out.println("任务被拒绝: " + r.toString());
        
        // 尝试重新提交
        try {
            if (!executor.getQueue().offer(r, 5, TimeUnit.SECONDS)) {
                // 最终处理: 记录到数据库或消息队列
                saveToDatabase(r);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            saveToDatabase(r);
        }
    }
    
    private void saveToDatabase(Runnable r) {
        // 保存到数据库,后续重试
    }
}

// 使用自定义拒绝策略
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 20, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new CustomRejectedExecutionHandler()
);

// 不同业务使用不同的线程池
public class ThreadPoolConfig {
    // 订单业务线程池
    public ThreadPoolExecutor orderPool = new ThreadPoolExecutor(
        10, 20, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadFactoryBuilder().setNameFormat("order-pool-%d").build()
    );
    
    // 支付业务线程池
    public ThreadPoolExecutor paymentPool = new ThreadPoolExecutor(
        5, 10, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(50),
        new ThreadFactoryBuilder().setNameFormat("payment-pool-%d").build()
    );
    
    // 通知业务线程池
    public ThreadPoolExecutor notificationPool = new ThreadPoolExecutor(
        3, 5, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(20),
        new ThreadFactoryBuilder().setNameFormat("notification-pool-%d").build()
    );
}

public class DynamicThreadPool {
    private ThreadPoolExecutor executor;
    
    public void adjustThreadPool(int corePoolSize, int maxPoolSize) {
        // 先调整最大线程数（必须大于等于核心线程数）
        if (maxPoolSize < executor.getCorePoolSize()) {
            executor.setMaximumPoolSize(executor.getCorePoolSize());
            executor.setCorePoolSize(corePoolSize);
            executor.setMaximumPoolSize(maxPoolSize);
        } else {
            executor.setCorePoolSize(corePoolSize);
            executor.setMaximumPoolSize(maxPoolSize);
        }
        
        System.out.println("线程池调整: corePoolSize=" + corePoolSize + ", maxPoolSize=" + maxPoolSize);
    }
    
    // 根据负载动态调整
    public void autoAdjust() {
        double usage = (double) executor.getActiveCount() / executor.getMaximumPoolSize();
        
        if (usage > 0.8 && executor.getQueue().size() > 50) {
            // 负载高,扩容
            int newCorePoolSize = Math.min(executor.getCorePoolSize() * 2, 50);
            int newMaxPoolSize = Math.min(executor.getMaximumPoolSize() * 2, 100);
            adjustThreadPool(newCorePoolSize, newMaxPoolSize);
        } else if (usage < 0.3) {
            // 负载低,缩容
            int newCorePoolSize = Math.max(executor.getCorePoolSize() / 2, 5);
            int newMaxPoolSize = Math.max(executor.getMaximumPoolSize() / 2, 10);
            adjustThreadPool(newCorePoolSize, newMaxPoolSize);
        }
    }
}

# 1. 找到 CPU 使用率高的进程
top
# 假设 PID 为 12345

# 2. 查看该进程中的线程
top -H -p 12345
# 发现 PID 为 12350、12351 的线程 CPU 使用率高

# 3. 将线程 PID 转换为十六进制
printf "%x\n" 12350
# 输出: 303e

# 4. 使用 jstack 查看线程堆栈
jstack 12345 | grep -A 20 303e

# 5. 分析线程堆栈

"http-nio-8080-exec-1" #25 daemon prio=5 os_prio=0 tid=0x00007f8a9c001000 nid=0x303e runnable [0x00007f8a8c1f8000]
   java.lang.Thread.State: RUNNABLE
        at com.example.UserService.calculateScore(UserService.java:123)
        at com.example.ScoreController.getScore(ScoreController.java:45)
        ...

public int calculateScore(Long userId) {
    int score = 0;
    while (true) {  // 死循环
        score += getUserScore(userId);
        if (score > 1000) {
            break;
        }
    }
    return score;
}

public int calculateScore(Long userId) {
    int score = 0;
    int retryCount = 0;
    int maxRetry = 10;  // 添加最大重试次数
    
    while (retryCount < maxRetry) {
        score += getUserScore(userId);
        if (score > 1000) {
            break;
        }
        retryCount++;
    }
    return score;
}

# 1. 使用 jstack 检测死锁
jstack -l <pid> | grep -A 10 "Found one Java-level deadlock"

Found one Java-level deadlock:
=============================
"Thread-1":
  waiting to lock monitor 0x00007f8a9c0034c8 (object 0x00000000e1a2a8c0, a com.example.Account),
  which is held by "Thread-2"
"Thread-2":
  waiting to lock monitor 0x00007f8a9c0032c8 (object 0x00000000e1a2a8d0, a com.example.Account),
  which is held by "Thread-1"

Java stack information for the threads listed above:
===================================================
"Thread-1":
        at com.example.TransferService.transfer(TransferService.java:25)
        - waiting to lock <0x00000000e1a2a8c0> (a com.example.Account)
        - locked <0x00000000e1a2a8d0> (a com.example.Account)
"Thread-2":
        at com.example.TransferService.transfer(TransferService.java:25)
        - waiting to lock <0x00000000e1a2a8d0> (a com.example.Account)
        - locked <0x00000000e1a2a8c0> (a com.example.Account)

public class TransferService {
    public void transfer(Account from, Account to, BigDecimal amount) {
        synchronized (from) {
            synchronized (to) {
                from.debit(amount);
                to.credit(amount);
            }
        }
    }
}

public void transfer(Account from, Account to, BigDecimal amount) {
    // 按 ID 排序,确保所有线程按相同顺序获取锁
    Account first = from.getId() < to.getId() ? from : to;
    Account second = from.getId() < to.getId() ? to : from;
    
    synchronized (first) {
        synchronized (second) {
            from.debit(amount);
            to.credit(amount);
        }
    }
}

private final Object globalLock = new Object();

public void transfer(Account from, Account to, BigDecimal amount) {
    synchronized (globalLock) {
        synchronized (from) {
            synchronized (to) {
                from.debit(amount);
                to.credit(amount);
            }
        }
    }
}

public void transfer(Account from, Account to, BigDecimal amount) {
    Lock fromLock = from.getLock();
    Lock toLock = to.getLock();
    
    while (true) {
        try {
            if (fromLock.tryLock(100, TimeUnit.MILLISECONDS)) {
                try {
                    if (toLock.tryLock(100, TimeUnit.MILLISECONDS)) {
                        try {
                            from.debit(amount);
                            to.credit(amount);
                            return;
                        } finally {
                            toLock.unlock();
                        }
                    }
                } finally {
                    fromLock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("转账被中断", e);
        }
    }
}

# 1. 使用 Arthas 查看线程池状态
thread -n 3

# 2. 查看线程堆栈
jstack <pid> | grep -A 20 "http-nio"

# 3. 使用 Arthas 监控方法执行时间
trace com.example.UserService getUser

// 线程池配置
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,  // 核心线程数
    10,  // 最大线程数
    0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<>(),  // 无界队列
    new ThreadPoolExecutor.AbortPolicy()
);

// 任务提交速度 > 处理速度
for (int i = 0; i < 100000; i++) {
    executor.submit(() -> {
        // 慢任务
        Thread.sleep(1000);
    });
}

// 1. 使用有界队列
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,
    20,  // 增加最大线程数
    60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),  // 有界队列
    new ThreadFactoryBuilder().setNameFormat("worker-%d").build(),
    new ThreadPoolExecutor.CallerRunsPolicy()  // 调用者执行策略
);

// 2. 添加监控
new Thread(() -> {
    while (true) {
        System.out.println("活跃线程: " + executor.getActiveCount());
        System.out.println("队列大小: " + executor.getQueue().size());
        if (executor.getQueue().size() > 80) {
            AlertService.sendAlert("线程池队列积压告警");
        }
        Thread.sleep(1000);
    }
}).start();

// 3. 添加熔断
if (executor.getQueue().size() > 90) {
    throw new RuntimeException("系统繁忙,请稍后重试");
}

# 1. 使用 jmap 查看堆内存使用
jmap -heap <pid>

# 2. 导出堆 dump
jmap -dump:format=b,file=heap.hprof <pid>

# 3. 使用 VisualVM 或 MAT 分析堆 dump

1. 打开 VisualVM,加载 heap.hprof
2. 点击"类"标签,按实例数排序
3. 发现 UserSession 实例数异常多
4. 右键点击 UserSession,选择"在堆中显示"
5. 查看引用链,找到 GC Root

public class SessionManager {
    private static final Map<String, UserSession> sessions = new HashMap<>();
    
    public void createSession(String sessionId, UserSession session) {
        sessions.put(sessionId, session);
    }
    
    // 没有 remove 方法,session 不会被清理
}

public class SessionManager {
    // 使用 ConcurrentHashMap + 定时清理
    private static final ConcurrentHashMap<String, UserSession> sessions = new ConcurrentHashMap<>();
    
    public void createSession(String sessionId, UserSession session) {
        sessions.put(sessionId, session);
    }
    
    public void removeSession(String sessionId) {
        sessions.remove(sessionId);
    }
    
    // 定时清理过期 session
    public void cleanExpiredSessions() {
        sessions.entrySet().removeIf(entry -> 
            entry.getValue().isExpired()
        );
    }
    
    // 使用 Guava Cache
    private static final Cache<String, UserSession> sessions = CacheBuilder.newBuilder()
        .expireAfterAccess(30, TimeUnit.MINUTES)
        .maximumSize(10000)
        .build();
}

# 1. 找到 CPU 使用率高的进程
top

# 2. 查看进程中的线程
top -H -p <pid>

# 3. 将线程 PID 转为十六进制
printf "%x\n" <tid>

# 4. 使用 jstack 查看
jstack <pid> | grep -A 20 <hex_tid>

# 1. 监控方法执行时间
trace com.example.UserService getUser

# 2. 查看方法调用参数和返回值
watch com.example.UserService getUser '{params, returnObj}' -x 2

# 3. 查看线程堆栈
thread -n 3

# 方法 1: jstack
jstack -l <pid> | grep -A 10 "Found one Java-level deadlock"

# 方法 2: jconsole
# 点击"线程"标签,点击"检测死锁"

# 方法 3: Arthas
thread -b

AtomicStampedReference<Integer> ref = new AtomicStampedReference<>(1, 0);

// 更新时检查版本
int[] stampHolder = new int[1];
Integer value = ref.get(stampHolder);
ref.compareAndSet(value, 2, stampHolder[0], stampHolder[0] + 1);

// 方法 1: 双重检查锁定
public class Singleton {
    private static volatile Singleton instance;
    
    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();
                }
            }
        }
        return instance;
    }
}

// 方法 2: 静态内部类
public class Singleton {
    private Singleton() {}
    
    private static class Holder {
        private static final Singleton INSTANCE = new Singleton();
    }
    
    public static Singleton getInstance() {
        return Holder.INSTANCE;
    }
}

// 方法 3: 枚举
public enum Singleton {
    INSTANCE;
    
    public void doSomething() {
        // ...
    }
}

// 方法 1: AtomicInteger
public class AtomicCounter {
    private AtomicInteger count = new AtomicInteger(0);
    
    public void increment() {
        count.incrementAndGet();
    }
    
    public int getCount() {
        return count.get();
    }
}

// 方法 2: LongAdder（高并发场景）
public class LongAdderCounter {
    private LongAdder count = new LongAdder();
    
    public void increment() {
        count.increment();
    }
    
    public long getCount() {
        return count.sum();
    }
}

// 方法 3: synchronized
public class SynchronizedCounter {
    private int count = 0;
    
    public synchronized void increment() {
        count++;
    }
    
    public synchronized int getCount() {
        return count;
    }
}

并发问题排查思路:
    1. 看现象
       ├── 数据不一致 → 竞态条件、原子性
       ├── RT 飙升 → 死锁、阻塞、线程池堆积
       ├── CPU 高但吞吐下降 → 锁竞争、资源争抢
       └── 内存持续增长 → 内存泄漏
    
    2. 找原因
       ├── jstack → 线程堆栈分析
       ├── jconsole/VisualVM → 图形化监控
       ├── Arthas → 动态诊断
       └── jmap → 内存分析
    
    3. 定位代码
       ├── 分析线程堆栈
       ├── 追踪方法调用
       ├── 查看对象引用链
       └── 分析锁竞争
    
    4. 修复问题
       ├── 加锁/优化锁
       ├── 使用并发工具
       ├── 调整线程池参数
       └── 修复内存泄漏

// × 危险代码
if (!orderRepository.exists(orderNo)) {
    orderRepository.save(new Order(orderNo));
}

// 安全代码
// 方法 1: 数据库唯一键
// 方法 2: 分布式锁
// 方法 3: 原子操作
// 方法 4: 状态机