java - ExecutorService 令人惊讶的性能盈亏平衡点——经验法则？

Question

我正在尝试了解如何正确使用 Java 的执行器。我意识到将任务提交给 ExecutorService 有其自身的开销。但是，我很惊讶地看到它竟然这么高。

我的程序需要以尽可能低的延迟处理大量数据(股票市场数据)。大多数计算都是相当简单的算术运算。

我试图测试一些非常简单的东西:“Math.random() * Math.random()”

最简单的测试在一个简单的循环中运行这个计算。第二个测试在匿名 Runnable 中进行相同的计算(这应该衡量创建新对象的成本)。第三个测试将 Runnable 传递给 ExecutorService(这衡量了引入执行程序的成本)。

我在我的小笔记本电脑(2 个 cpu，1.5 gig ram)上运行了测试:

(in milliseconds)
simpleCompuation:47
computationWithObjCreation:62
computationWithObjCreationAndExecutors:422

(大约每四次运行一次，前两个数字最终相等)

请注意，执行器比在单个线程上执行花费的时间要多得多。对于 1 到 8 之间的线程池大小，数字大致相同。

问题:我是否漏掉了一些明显的东西，或者这些结果是否符合预期？这些结果告诉我，我传递给执行者的任何任务都必须进行一些重要的计算。如果我正在处理数百万条消息，并且我需要对每条消息执行非常简单(且成本低廉)的转换，我仍然可能无法使用执行器......尝试将计算分散到多个 CPU 上最终可能比仅在一个线程中完成它们。设计决策变得比我原先想象的要复杂得多。有什么想法吗？

---

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ExecServicePerformance {

 private static int count = 100000;

 public static void main(String[] args) throws InterruptedException {

  //warmup
  simpleCompuation();
  computationWithObjCreation();
  computationWithObjCreationAndExecutors();

  long start = System.currentTimeMillis();
  simpleCompuation();
  long stop = System.currentTimeMillis();
  System.out.println("simpleCompuation:"+(stop-start));

  start = System.currentTimeMillis();
  computationWithObjCreation();
  stop = System.currentTimeMillis();
  System.out.println("computationWithObjCreation:"+(stop-start));

  start = System.currentTimeMillis();
  computationWithObjCreationAndExecutors();
  stop = System.currentTimeMillis();
  System.out.println("computationWithObjCreationAndExecutors:"+(stop-start));


 }

 private static void computationWithObjCreation() {
  for(int i=0;i<count;i++){
   new Runnable(){

    @Override
    public void run() {
     double x = Math.random()*Math.random();
    }

   }.run();
  }

 }

 private static void simpleCompuation() {
  for(int i=0;i<count;i++){
   double x = Math.random()*Math.random();
  }

 }

 private static void computationWithObjCreationAndExecutors()
   throws InterruptedException {

  ExecutorService es = Executors.newFixedThreadPool(1);
  for(int i=0;i<count;i++){
   es.submit(new Runnable() {
    @Override
    public void run() {
     double x = Math.random()*Math.random();     
    }
   });
  }
  es.shutdown();
  es.awaitTermination(10, TimeUnit.SECONDS);
 }
}

Answer 1

1. 使用执行器是关于利用 CPU 和/或 CPU 核心，因此如果您创建一个线程池以最大程度地利用 CPU 数量，则您必须拥有与 CPU/核心一样多的线程。
2. 你是对的，创建新对象的成本太高了。因此，减少开支的一种方法是使用批处理。如果您知道要进行的计算的种类和数量，就可以创建批处理。因此，请考虑在一项已执行任务中完成的数千次计算。您为每个线程创建批处理。一旦计算完成 (java.util.concurrent.Future)，您就可以创建下一批。甚至可以并行创建新批处理(4 个 CPU -> 3 个线程用于计算，1 个线程用于批处理配置)。最后，您可能会获得更高的吞吐量，但内存需求更高(批处理、供应)。

编辑:我更改了您的示例，让它在我的小型双核 x200 笔记本电脑上运行。

provisioned 2 batches to be executed
simpleCompuation:14
computationWithObjCreation:17
computationWithObjCreationAndExecutors:9

如您在源代码中所见，我也将批处理配置和执行程序生命周期排除在测量之外。与其他两种方法相比，这更公平。

自己看结果...

import java.util.List;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ExecServicePerformance {

    private static int count = 100000;

    public static void main( String[] args ) throws InterruptedException {

        final int cpus = Runtime.getRuntime().availableProcessors();

        final ExecutorService es = Executors.newFixedThreadPool( cpus );

        final Vector< Batch > batches = new Vector< Batch >( cpus );

        final int batchComputations = count / cpus;

        for ( int i = 0; i < cpus; i++ ) {
            batches.add( new Batch( batchComputations ) );
        }

        System.out.println( "provisioned " + cpus + " batches to be executed" );

        // warmup
        simpleCompuation();
        computationWithObjCreation();
        computationWithObjCreationAndExecutors( es, batches );

        long start = System.currentTimeMillis();
        simpleCompuation();
        long stop = System.currentTimeMillis();
        System.out.println( "simpleCompuation:" + ( stop - start ) );

        start = System.currentTimeMillis();
        computationWithObjCreation();
        stop = System.currentTimeMillis();
        System.out.println( "computationWithObjCreation:" + ( stop - start ) );

        // Executor

        start = System.currentTimeMillis();
        computationWithObjCreationAndExecutors( es, batches );    
        es.shutdown();
        es.awaitTermination( 10, TimeUnit.SECONDS );
        // Note: Executor#shutdown() and Executor#awaitTermination() requires
        // some extra time. But the result should still be clear.
        stop = System.currentTimeMillis();
        System.out.println( "computationWithObjCreationAndExecutors:"
                + ( stop - start ) );
    }

    private static void computationWithObjCreation() {

        for ( int i = 0; i < count; i++ ) {
            new Runnable() {

                @Override
                public void run() {

                    double x = Math.random() * Math.random();
                }

            }.run();
        }

    }

    private static void simpleCompuation() {

        for ( int i = 0; i < count; i++ ) {
            double x = Math.random() * Math.random();
        }

    }

    private static void computationWithObjCreationAndExecutors(
            ExecutorService es, List< Batch > batches )
            throws InterruptedException {

        for ( Batch batch : batches ) {
            es.submit( batch );
        }

    }

    private static class Batch implements Runnable {

        private final int computations;

        public Batch( final int computations ) {

            this.computations = computations;
        }

        @Override
        public void run() {

            int countdown = computations;
            while ( countdown-- > -1 ) {
                double x = Math.random() * Math.random();
            }
        }
    }
}