100%CPU使用率的Java线程优化

Question

我有一个应用程序接受队列上的工作,然后旋转,在独立的线程上完成.线程数量不大,比如高达100,但这些是密集型任务,可以快速将CPU提升到100%.

为了最快地完成大部分工作:我最好在我需要做更多工作时启动更多线程,让Java线程调度程序处理分配工作,或者更聪明地管理工作负载以保持CPU低于100%让我更快？

这台机器专用于我的java应用程序.

编辑:

感谢您的精彩输入!

这些任务具有不同的复杂性并且涉及I/O,因此具有较低的4个线程池可能只是将CPU运行到20%.我无法知道有多少任务实际上将CPU带到100%.

我的想法是我应该通过RMI监控CPU并动态拨打上下工作,或者我应该不关心并让操作系统处理它.

Answer 1

如果在并行线程中有太多同时执行计算密集型任务,则会很快达到收益递减的程度.实际上,如果有N个处理器(核心),那么你不需要超过N个这样的线程.现在,如果任务偶尔暂停I/O或用户交互,那么正确的数字可能会稍大一些.但在一般情况下,如果在任何一个时刻有多个线程想要做的计算比可用的核心,那么你的程序是在浪费时间上下文切换 - 即,调度成本你.

Answer 2

为了最快地完成大部分工作:我最好在我需要做更多工作时启动更多线程,让Java线程调度程序处理分配工作,或者更聪明地管理工作负载以保持CPU低于100%让我更快？

随着您添加越来越多的线程,上下文切换,内存缓存刷新,内存缓存溢出以及内核和JVM线程管理所产生的开销也会增加.当您的线程占用CPU时,它们的内核优先级降至最低,并且它们将达到时间片最小值.随着越来越多的线程占用内存,它们会溢出各种内部CPU内存缓存.CPU需要从较慢的内存中交换作业的可能性更高.JVM内部存在更多互斥锁本地争用,可能还有一些(可能很小的)增量每线程和对象带宽GC开销.根据用户任务的同步程度,更多线程会导致内存刷新和锁争用增加.

对于任何程序和任何体系结构,线程都可以最佳地利用可用的处理器和IO资源,同时限制内核和JVM开销.反复寻找最佳位置将需要多次迭代和一些猜测.

我建议使用Executors.newFixedThreadPool(SOME_NUMBER);并提交你的工作.然后,您可以执行多次运行,根据工作和框架的结构,上下调整线程数,直到找到同时运行的最佳池数.

但是要理解,最佳线程数将根据处理器的数量和可能非常重要的其他因素而变化.如果它们阻塞磁盘或网络IO资源,则可能需要更多线程.如果他们正在做的工作主要是基于CPU的,那么线程就会减少.

Answer 3

你的CPU以100%运行的事实并没有说明他们在做有用工作的忙碌程度.在您的情况下,您使用的线程数多于核心数,因此100%包含一些上下文切换并且不必要地使用内存(对100个线程影响很小),这是次优的.

对于CPU密集型任务,我通常使用这个成语:

private final int NUM_THREADS = Runtime.getRuntime().availableProcessors() + 1;
private final ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);

正如其他人所指出的,使用更多线程只会引入不必要的上下文切换.

显然,如果任务执行某些I/O和其他阻塞操作,则这不适用,并且更大的池将有意义.

编辑

要回复@MartinJames评论,我已经运行了一个(简单化)基准测试 - 结果显示,从池大小=处理器数量+ 1到100只会略微降低性能(让我们称之为5%) - 更高的数字( 1000和10000)确实显着地达到了性能.

结果是10次运行的平均值:
池大小:9:238毫秒.//(NUM_CORES + 1)
池大小:100:245毫秒.
泳池大小:1000:319毫秒.
池大小:10000:2482毫秒.

码:

public class Test {

    private final static int NUM_CORES = Runtime.getRuntime().availableProcessors();
    private static long count;
    private static Runnable r = new Runnable() {

        @Override
        public void run() {
            int count = 0;
            for (int i = 0; i < 100_000; i++) {
                count += i;
            }
            Test.count += count;
        }
    };

    public static void main(String[] args) throws Exception {
        //warmup
        runWith(10);

        //test
        runWith(NUM_CORES + 1);
        runWith(100);
        runWith(1000);
        runWith(10000);
    }

    private static void runWith(int poolSize) throws InterruptedException {
        long average = 0;
        for (int run = 0; run < 10; run++) { //run 10 times and take the average
            Test.count = 0;
            ExecutorService executor = Executors.newFixedThreadPool(poolSize);
            long start = System.nanoTime();
            for (int i = 0; i < 50000; i++) {
                executor.submit(r);
            }
            executor.shutdown();
            executor.awaitTermination(10, TimeUnit.SECONDS);
            long end = System.nanoTime();
            average += ((end - start) / 1000000);
            System.gc();
        }
        System.out.println("Pool size: " + poolSize + ": " + average / 10 + " ms.  ");
    }
}