Java 线程池配置指南

1
public class CustomThreadFactory implements ThreadFactory {
2
    private final String prefix;
3
    private final AtomicInteger counter = new AtomicInteger(1);
4

5
    public CustomThreadFactory(String prefix) {
6
        this.prefix = prefix;
7
    }
8

9
    @Override
10
    public Thread newThread(Runnable r) {
11
        Thread t = new Thread(r, prefix + "-" + counter.getAndIncrement());
12
        t.setDaemon(false);
13
        t.setUncaughtExceptionHandler((thread, ex) ->
14
            log.error("线程 {} 发生未捕获异常", thread.getName(), ex));
15
        return t;
16
    }
17
}

1
// 问题版本
2
int cores = Runtime.getRuntime().availableProcessors();
3

4
// 安全版本：读取容器 CPU 配额
5
private int getCpuCores() {
6
    String cpuLimit = System.getenv("CPU_LIMIT");
7
    if (cpuLimit != null) return Integer.parseInt(cpuLimit);
8

9
    try {
10
        Path quotaPath = Paths.get("/sys/fs/cgroup/cpu/cpu.cfs_quota_us");
11
        Path periodPath = Paths.get("/sys/fs/cgroup/cpu/cpu.cfs_period_us");
12
        long quota = Long.parseLong(Files.readString(quotaPath).trim());
13
        long period = Long.parseLong(Files.readString(periodPath).trim());
14
        if (quota > 0) return (int) Math.ceil((double) quota / period);
15
    } catch (Exception ignored) {}
16

17
    return Runtime.getRuntime().availableProcessors();
18
}

1
@Bean("orderExecutor")
2
public ThreadPoolExecutor orderExecutor() {
3
    int cores = Runtime.getRuntime().availableProcessors();
4
    int effectiveCores = Math.min(cores, Integer.parseInt(
5
        System.getenv().getOrDefault("CPU_LIMIT", String.valueOf(cores))));
6

7
    int dbPoolSize = 50;
8
    int coreSize = Math.min(effectiveCores * 10, dbPoolSize);
9
    int maxSize = (int) (coreSize * 1.2);
10
    int queueCap = coreSize * 6;
11

12
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
13
        coreSize, maxSize,
14
        60L, TimeUnit.SECONDS,
15
        new LinkedBlockingQueue<>(queueCap),
16
        new CustomThreadFactory("order-exec"),
17
        new MetricsRejectedHandler("order")
18
    );
19
    executor.allowCoreThreadTimeOut(false);
20
    return executor;
21
}

1
public class MetricsRejectedHandler implements RejectedExecutionHandler {
2
    private final String poolName;
3

4
    public MetricsRejectedHandler(String poolName) {
5
        this.poolName = poolName;
6
    }
7

8
    @Override
9
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
10
        Metrics.counter("threadpool.rejected", "pool", poolName).increment();
11
        log.error("[{}] 任务被拒绝，队列积压:{} 活跃线程:{} 最大线程:{}",
12
            poolName, e.getQueue().size(), e.getActiveCount(), e.getMaximumPoolSize());
13
        throw new RejectedExecutionException("Pool " + poolName + " is saturated");
14
    }
15
}

1
@Component
2
public class DynamicThreadPoolManager {
3

4
    @Autowired
5
    private ThreadPoolExecutor orderExecutor;
6

7
    @NacosConfigListener(dataId = "thread-pool-config", groupId = "DEFAULT_GROUP")
8
    public void onConfigChange(String configJson) {
9
        ThreadPoolConfig cfg = JSON.parseObject(configJson, ThreadPoolConfig.class);
10

11
        int newCore = cfg.getCoreSize();
12
        int newMax = cfg.getMaxSize();
13

14
        // 扩大时先改 max，缩小时先改 core
15
        if (newCore > orderExecutor.getMaximumPoolSize()) {
16
            orderExecutor.setMaximumPoolSize(newMax);
17
            orderExecutor.setCorePoolSize(newCore);
18
        } else {
19
            orderExecutor.setCorePoolSize(newCore);
20
            orderExecutor.setMaximumPoolSize(newMax);
21
        }
22

23
        log.info("线程池动态调整完成 pool=order core={} max={}", newCore, newMax);
24
    }
25
}

1
public static void registerToPrometheus(ThreadPoolExecutor pool, String name) {
2
    MeterRegistry registry = Metrics.globalRegistry;
3

4
    Gauge.builder("threadpool.active", pool, ThreadPoolExecutor::getActiveCount)
5
         .tag("pool", name).description("活跃线程数").register(registry);
6

7
    Gauge.builder("threadpool.pool_size", pool, ThreadPoolExecutor::getPoolSize)
8
         .tag("pool", name).description("当前线程总数").register(registry);
9

10
    Gauge.builder("threadpool.queue_size", pool, p -> p.getQueue().size())
11
         .tag("pool", name).description("队列积压任务数").register(registry);
12

13
    Gauge.builder("threadpool.utilization", pool,
14
             p -> (double) p.getActiveCount() / p.getCorePoolSize())
15
         .tag("pool", name).description("核心线程利用率").register(registry);
16

17
    Gauge.builder("threadpool.largest_pool_size", pool, ThreadPoolExecutor::getLargestPoolSize)
18
         .tag("pool", name).description("历史最大线程数").register(registry);
19
}

1
groups:
2
  - name: threadpool_alerts
3
    rules:
4
      - alert: ThreadPoolHighUtilization
5
        expr: threadpool_utilization > 0.85
6
        for: 2m
7
        labels:
8
          severity: warning
9
        annotations:
10
          summary: "线程池 {{ $labels.pool }} 利用率超过 85%"
11

12
      - alert: ThreadPoolQueueBacklog
13
        expr: threadpool_queue_size / threadpool_queue_capacity > 0.7
14
        for: 1m
15
        labels:
16
          severity: warning
17
        annotations:
18
          summary: "线程池 {{ $labels.pool }} 队列积压超过 70%"
19

20
      - alert: ThreadPoolRejection
21
        expr: increase(threadpool_rejected_total[1m]) > 0
22
        for: 0m
23
        labels:
24
          severity: critical
25
        annotations:
26
          summary: "线程池 {{ $labels.pool }} 出现任务拒绝"
27

28
      - alert: ThreadPoolMaxSizeReached
29
        expr: threadpool_pool_size >= threadpool_max_size
30
        for: 3m
31
        labels:
32
          severity: warning
33
        annotations:
34
          summary: "线程池 {{ $labels.pool }} 线程数触达最大值，持续 3 分钟"