ejemplo - java threadpoolexecutor

El sorprendente punto de equilibrio de rendimiento de ExecutorService: ¿reglas básicas?

Question

Estoy tratando de averiguar cómo usar correctamente los Ejecutores de Java. Me doy cuenta de que enviar tareas a un ExecutorService tiene su propia sobrecarga. Sin embargo, me sorprende ver que es tan alto como es.

Mi programa necesita procesar una gran cantidad de datos (datos del mercado de valores) con la menor latencia posible. La mayoría de los cálculos son operaciones aritméticas bastante simples.

Intenté probar algo muy simple: " Math.random() * Math.random() "

La prueba más simple ejecuta este cálculo en un bucle simple. La segunda prueba realiza el mismo cálculo dentro de un Runnable anónimo (se supone que esto mide el costo de crear nuevos objetos). La tercera prueba pasa el Runnable a un ExecutorService (esto mide el costo de introducción de ejecutores).

Corrí las pruebas en mi portátil dinky (2 CPU, 1.5 gig ram):

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

(aproximadamente una vez de cada cuatro carreras, los dos primeros números terminan siendo iguales)

Tenga en cuenta que los ejecutores toman mucho, mucho más tiempo que ejecutando en un solo hilo. Los números fueron aproximadamente los mismos para tamaños de grupo de hilos entre 1 y 8.

Pregunta: ¿Me estoy perdiendo algo obvio o se esperan estos resultados? Estos resultados me dicen que cualquier tarea que pase a un ejecutor debe realizar algún cálculo no trivial. Si estoy procesando millones de mensajes, y necesito realizar transformaciones muy simples (y baratas) en cada mensaje, es posible que todavía no pueda usar ejecutores ... al tratar de distribuir los cálculos en varias CPU puede ser más costoso que simplemente Haciéndolos en un solo hilo. La decisión de diseño se vuelve mucho más compleja de lo que originalmente pensé. ¿Alguna idea?

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. El uso de ejecutores consiste en utilizar CPU y / o núcleos de CPU, por lo que si crea un grupo de subprocesos que utiliza la cantidad de CPU en el mejor de los casos, debe tener tantos subprocesos como CPU / núcleos.
2. Tienes razón, crear nuevos objetos cuesta demasiado. Así que una forma de reducir los gastos es usar lotes. Si conoce el tipo y la cantidad de cálculos que debe hacer, crea lotes. Entonces piense en miles de cálculos realizados en una tarea ejecutada. Se crean lotes para cada hilo. Tan pronto como se realiza el cálculo (java.util.concurrent.Future), crea el siguiente lote. Incluso la creación de nuevos lotes se puede hacer en parralel (4 CPU -> 3 hilos para computación, 1 hilo para aprovisionamiento por lotes). Al final, puede terminar con más rendimiento, pero con mayores demandas de memoria (lotes, aprovisionamiento).

Edición: cambié su ejemplo y lo dejé correr en mi pequeña computadora portátil x200 de doble núcleo.

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

Como se ve en el código fuente, también eliminé de la medición el aprovisionamiento por lotes y el ciclo de vida del ejecutor. Eso es más justo en comparación con los otros dos métodos.

Mira los resultados por ti mismo ...

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();
            }
        }
    }
}

Answer 2

Aquí están los resultados en mi máquina (OpenJDK 8 en Ubuntu 14.0 de 64 bits, Thinkpad W530)

simpleCompuation:6
computationWithObjCreation:5
computationWithObjCreationAndExecutors:33

Ciertamente hay gastos generales. Pero recuerde cuáles son estos números: milisegundos para iteraciones de 100k . En su caso, la sobrecarga fue de aproximadamente 4 microsegundos por iteración. Para mí, la sobrecarga fue de alrededor de un cuarto de microsegundo.

La sobrecarga es la sincronización, las estructuras de datos internas y, posiblemente, la falta de optimización de JIT debido a las rutas de código complejas (ciertamente más complejas que su bucle for).

Las tareas que realmente desea paralelizar valdrían la pena, a pesar de la sobrecarga de un cuarto de microsegundo.

Para su información, este sería un muy mal cálculo para paralelizar. Subí el hilo a 8 (el número de núcleos):

simpleCompuation:5
computationWithObjCreation:6
computationWithObjCreationAndExecutors:38

No lo hizo más rápido. Esto se debe a que Math.random() está sincronizado.

Answer 3

El objetivo principal del Fixed ThreadPool es reutilizar los hilos ya creados. Por lo tanto, las mejoras de rendimiento se ven en la falta de la necesidad de volver a crear un nuevo hilo cada vez que se envía una tarea. Por lo tanto, el tiempo de parada debe tomarse dentro de la tarea enviada. Sólo con en la última declaración del método de ejecución.

Answer 4

En primer lugar hay algunos problemas con el microbenchmark. Haces un calentamiento, lo cual es bueno. Sin embargo, es mejor ejecutar la prueba varias veces, lo que debería dar una idea de si realmente se ha calentado y la varianza de los resultados. También tiende a ser mejor hacer la prueba de cada algoritmo en ejecuciones separadas, de lo contrario, podría causar la desoptimización cuando cambia un algoritmo.

La tarea es muy pequeña, aunque no estoy del todo seguro de cuán pequeña. Así que el número de veces más rápido es bastante sin sentido. En situaciones de subprocesos múltiples, tocará las mismas ubicaciones volátiles, por lo que los subprocesos podrían causar un rendimiento realmente malo (use una instancia Random por subproceso). También una carrera de 47 milisegundos es un poco corta.

Ciertamente, ir a otro hilo para una pequeña operación no va a ser rápido. Dividir las tareas en tamaños más grandes si es posible. JDK7 parece como si tuviera un marco de unión de bifurcación, que intenta admitir tareas finas de los algoritmos de dividir y conquistar prefiriendo ejecutar tareas en el mismo hilo en orden, con tareas más grandes extraídas por hilos inactivos.

Answer 5

Es necesario agrupar la ejecución de alguna manera, para enviar porciones más grandes de cómputo a cada hilo (por ejemplo, crear grupos basados ​​en el símbolo de valores) Obtuve los mejores resultados en escenarios similares utilizando el Disruptor. Tiene una sobrecarga por trabajo muy baja. Aún así, es importante para los trabajos grupales, el turno rotativo ingenuo por lo general crea muchos errores de caché.

consulte http://java-is-the-new-c.blogspot.de/2014/01/comparision-of-different-concurrency.html

Answer 6

Esta no es una prueba justa para el grupo de hilos por las siguientes razones,

1. No estás aprovechando la agrupación en absoluto porque solo tienes 1 hilo.
2. El trabajo es demasiado simple como para que la sobrecarga de la agrupación no pueda justificarse. Una multiplicación en una CPU con FPP solo toma unos pocos ciclos.

Teniendo en cuenta los siguientes pasos adicionales que debe realizar el grupo de hilos además de la creación de objetos y la ejecución del trabajo,

1. Pon el trabajo en la cola
2. Eliminar el trabajo de la cola
3. Obtener el hilo de la agrupación y ejecutar el trabajo
4. Devuelve el hilo a la piscina.

Cuando tiene un trabajo real y varios subprocesos, el beneficio del conjunto de subprocesos será evidente.

Answer 7

La ''sobrecarga'' que menciona no tiene nada que ver con ExecutorService, se debe a la sincronización de múltiples subprocesos en Math.random, lo que crea una contención de bloqueo.

Entonces sí, te estás perdiendo algo (y la respuesta "correcta" a continuación no es realmente correcta).

Aquí hay un código de Java 8 para demostrar 8 subprocesos que ejecutan una función simple en la que no hay contención de bloqueo:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.DoubleFunction;

import com.google.common.base.Stopwatch;

public class ExecServicePerformance {

    private static final int repetitions = 120;
    private static int totalOperations = 250000;
    private static final int cpus = 8;
    private static final List<Batch> batches = batches(cpus);

    private static DoubleFunction<Double> performanceFunc = (double i) -> {return Math.sin(i * 100000 / Math.PI); };

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

        printExecutionTime("Synchronous", ExecServicePerformance::synchronous);
        printExecutionTime("Synchronous batches", ExecServicePerformance::synchronousBatches);
        printExecutionTime("Thread per batch", ExecServicePerformance::asynchronousBatches);
        printExecutionTime("Executor pool", ExecServicePerformance::executorPool);

    }

    private static void printExecutionTime(String msg, Runnable f) throws InterruptedException {
        long time = 0;
        for (int i = 0; i < repetitions; i++) {
            Stopwatch stopwatch = Stopwatch.createStarted();
            f.run(); //remember, this is a single-threaded synchronous execution since there is no explicit new thread
            time += stopwatch.elapsed(TimeUnit.MILLISECONDS);
        }
        System.out.println(msg + " exec time: " + time);
    }    

    private static void synchronous() {
        for ( int i = 0; i < totalOperations; i++ ) {
            performanceFunc.apply(i);
        }
    }

    private static void synchronousBatches() {      
        for ( Batch batch : batches) {
            batch.synchronously();
        }
    }

    private static void asynchronousBatches() {

        CountDownLatch cb = new CountDownLatch(cpus);

        for ( Batch batch : batches) {
            Runnable r = () ->  { batch.synchronously(); cb.countDown(); };
            Thread t = new Thread(r);
            t.start();
        }

        try {
            cb.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }        
    }

    private static void executorPool() {

        final ExecutorService es = Executors.newFixedThreadPool(cpus);

        for ( Batch batch : batches ) {
            Runnable r = () ->  { batch.synchronously(); };
            es.submit(r);
        }

        es.shutdown();

        try {
            es.awaitTermination( 10, TimeUnit.SECONDS );
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } 

    }

    private static List<Batch> batches(final int cpus) {
        List<Batch> list = new ArrayList<Batch>();
        for ( int i = 0; i < cpus; i++ ) {
            list.add( new Batch( totalOperations / cpus ) );
        }
        System.out.println("Batches: " + list.size());
        return list;
    }

    private static class Batch {

        private final int operationsInBatch;

        public Batch( final int ops ) {
            this.operationsInBatch = ops;
        }

        public void synchronously() {
            for ( int i = 0; i < operationsInBatch; i++ ) {
                performanceFunc.apply(i);
            }
        }
    }


}

Tiempos de resultados para 120 pruebas de 25k operaciones (ms):

• Tiempo de ejecución síncrono: 9956
• Tiempo de ejecución de lotes síncronos: 9900
• Hilo por tiempo de ejecución de lote: 2176
• Tiempo ejecutiva de la piscina ejecutora: 1922

Ganador: Servicio Ejecutor.

Answer 8

Math.random () en realidad se sincroniza en un solo generador de números aleatorios. Llamar a Math.random () da como resultado una contención significativa para el generador de números. De hecho, mientras más hilos tengas, más lento será.

Desde el Math.random () javadoc:

Este método está correctamente sincronizado para permitir el uso correcto por más de un hilo. Sin embargo, si muchos subprocesos necesitan generar números pseudoaleatorios a una gran velocidad, puede reducir la contención para que cada subproceso tenga su propio generador de números pseudoaleatorios.

Answer 9

No creo que esto sea en absoluto realista, ya que estás creando un nuevo servicio de ejecutor cada vez que realizas la llamada de método. A menos que tenga requisitos muy extraños que parezcan poco realistas, normalmente creará el servicio cuando se inicie la aplicación y luego le enviará trabajos.

Si vuelve a intentar la evaluación comparativa pero inicializa el servicio como un campo , una vez, fuera del ciclo de tiempo; luego verá la sobrecarga real de enviar Runnables al servicio en lugar de ejecutarlos usted mismo.

Pero no creo que hayas entendido bien el tema: los ejecutores no están destinados a estar allí por eficiencia, están allí para hacer que la coordinación y la entrega del trabajo a un grupo de subprocesos sea más simple. Siempre serán menos eficientes que solo invocar Runnable.run() usted mismo (ya que al final del día, el servicio ejecutor aún debe hacer esto, después de hacer un poco de limpieza adicional de antemano). Es cuando los estás usando desde múltiples hilos que necesitan procesamiento asíncrono, que realmente brillan.

También tenga en cuenta que está considerando la diferencia de tiempo relativa de un costo básicamente fijo (la sobrecarga del Ejecutor es la misma ya sea que sus tareas demoren 1 ms o 1 hora) en comparación con una cantidad variable muy pequeña (su trivial ejecutable). Si el servicio del ejecutor toma 5ms extra para ejecutar una tarea de 1ms, esa no es una cifra muy favorable. Si se necesitan 5 ms adicionales para ejecutar una tarea de 5 segundos (por ejemplo, una consulta de SQL no trivial), eso es completamente insignificante y vale la pena.

Entonces, hasta cierto punto, depende de su situación: si tiene una sección extremadamente crítica en el tiempo, que ejecuta muchas tareas pequeñas, que no es necesario ejecutar en paralelo o asincrónicamente, no obtendrá nada de un Ejecutor. Si está procesando tareas más pesadas en paralelo y desea responder de forma asíncrona (por ejemplo, una aplicación web), los Ejecutores son excelentes.

Si son la mejor opción para usted, depende de su situación, pero realmente necesita probar las pruebas con datos representativos realistas. No creo que sea apropiado sacar conclusiones de las pruebas que ha realizado, a menos que sus tareas sean realmente triviales (y no desee reutilizar la instancia del ejecutor ...).