根据 CPU 和 RAM 使用情况调整线程池的 Java Executor

Question

我的应用程序使用一个 Executor 为大量任务提供线程池。通过分析和基准测试，我确定当每个内核有多个线程时，我的应用程序运行速度最快。一个好的启发式方法是从每个内核 4 个线程开始，不断变化，直到达到 >90% 的 CPU 或 >90% 的 RAM。

是否有可用的执行器可以开箱即用？是自动为每个核心使用 N 个线程(而不是一个)，还是理想情况下，根据 CPU 和 RAM 使用情况限制线程池大小？

否则 - 如何以编程方式确定核心数？

Answer 1

一种方法是使用 ThreadPoolExecutor用core size共 1 个，起始 maximum pool size 4，然后根据内存和 CPU 使用率动态调整最大池大小。

恕我直言，更大的问题是如何测量内存使用和 CPU 负载。内存使用很简单:

public double memUsageRatio() {
  Runtime r = Runtime.getRuntime();
  return (double) (r.totalMemory() - r.freeMemory()) / r.maxMemory();
}

对于 CPU 负载，问题可能更大，具体取决于您运行的平台。在 Linux 上，您可以使用:

ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();

这将返回最后一分钟的系统平均负载。不幸的是，在 Windows 上，此方法总是返回 -1。过去，我通过计算所有线程的 CPU 时间总和除以所有处理器的所有运行时间总和，将其替换为给定时间间隔内系统平均负载的近似值。这只是一个近似值，但在大多数情况下效果很好:

import java.lang.management.*;

public class CPUUsageCollector implements Runnable {
  private final static long INTERVAL = 1000L; // polling interval in ms
  private long totalCpuTime = 0L; // total CPU time in millis
  private double load = 0d; // average load over the interval
  ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
  boolean stopped = false;

  @Override
  public void run() {
    try {
      while (!isStopped()) {
        long start = System.currentTimeMillis();
        long[] ids = threadMXBean.getAllThreadIds();
        long time = 0L;
        for (long id: ids) {
          long l = threadMXBean.getThreadCpuTime(id);
          if (l >= 0L) time += l;
        }
        long newCpuTime = time / 1000000L;
        synchronized(this) {
          long oldCpuTime = totalCpuTime;
          totalCpuTime = newCpuTime;
          // load = CPU time difference / sum of elapsed time for all CPUs
          load = (double) (newCpuTime - oldCpuTime) / 
           (double) (INTERVAL * Runtime.getRuntime().availableProcessors());
        }
        long sleepTime = INTERVAL - (System.currentTimeMillis() - start);
        goToSleep(sleepTime <= 0L ? INTERVAL : sleepTime);
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  public synchronized double getLoad() {
    return load;
  }

  public synchronized void goToSleep(final long time) {
    try {
      wait(time);
    } catch(InterruptedException e) {
      e.printStackTrace();
    }
  }

  public synchronized boolean isStopped() {
    return stopped;
  }

  public synchronized void setStopped(final boolean stopped) {
    this.stopped = stopped;
  }
}