Java Language
Ejecutor, ExecutorService y grupos de subprocesos

Introducción

Observaciones

Fuego y olvido - Tareas ejecutables

Los ejecutores aceptan un java.lang.Runnable que contiene código (potencialmente computacional o de otro modo pesado o pesado) para ejecutarse en otro Thread.

El uso sería:

Executor exec = anExecutor;
exec.execute(new Runnable() {
    @Override public void run() {
        //offloaded work, no need to get result back
    }
});

Tenga en cuenta que con este ejecutor, no tiene medios para recuperar ningún valor calculado.
Con Java 8, uno puede utilizar lambdas para acortar el ejemplo de código.

Executor exec = anExecutor;
exec.execute(() -> {
    //offloaded work, no need to get result back
});

ThreadPoolExecutor

Un Ejecutor común que se usa es el ThreadPoolExecutor , que se encarga del manejo de Thread. Puede configurar la cantidad mínima de subprocesos que el ejecutor siempre debe mantener cuando no hay mucho que hacer (se denomina tamaño del núcleo) y un tamaño máximo de subprocesos en el que la agrupación puede crecer, si hay más trabajo por hacer. Una vez que la carga de trabajo disminuye, el Grupo reduce lentamente el número de subprocesos nuevamente hasta que alcanza el tamaño mínimo.

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    1,                                     // keep at least one thread ready, 
                                           // even if no Runnables are executed
    5,                                     // at most five Runnables/Threads
                                           // executed in parallel
    1, TimeUnit.MINUTES,                   // idle Threads terminated after one
                                           // minute, when min Pool size exceeded
    new ArrayBlockingQueue<Runnable>(10)); // outstanding Runnables are kept here

pool.execute(new Runnable() {
    @Override public void run() {
        //code to run
    }
});

Nota: si configura ThreadPoolExecutor con una cola ilimitada , entonces el recuento de subprocesos no excederá corePoolSize ya que los nuevos subprocesos solo se crean si la cola está llena:

ThreadPoolExecutor con todos los parámetros:

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, 
TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler)

de JavaDoc

Si hay más de corePoolSize pero menos de maximumPoolSize se están ejecutando, se creará un nuevo thread solo si la cola está llena.

Ventajas:

1. El tamaño de BlockingQueue se puede controlar y los escenarios de memoria insuficiente se pueden evitar. El rendimiento de la aplicación no se degradará con el tamaño limitado de la cola delimitada.

2. Puede usar políticas existentes o crear nuevas políticas de manejo de rechazo.

   1. En el ThreadPoolExecutor.AbortPolicy predeterminado, el controlador lanza una excepción RejectedExecutionException en tiempo de ejecución cuando se rechaza.

   2. En ThreadPoolExecutor.CallerRunsPolicy , el hilo que invoca a sí mismo ejecuta la tarea. Esto proporciona un mecanismo de control de retroalimentación simple que reducirá la velocidad a la que se envían las nuevas tareas.

   3. En ThreadPoolExecutor.DiscardPolicy , una tarea que no se puede ejecutar simplemente se elimina.

   4. En ThreadPoolExecutor.DiscardOldestPolicy , si el ejecutor no se apaga, la tarea en la cabecera de la cola de trabajo se descarta, y luego se reintenta la ejecución (lo que puede fallar nuevamente, lo que hace que se repita).

3. ThreadFactory puede configurar ThreadFactory personalizado, lo cual es útil:

   1. Para establecer un nombre de hilo más descriptivo
   2. Para establecer el estado del daemon de hilo
   3. Para establecer la prioridad del hilo

Aquí hay un ejemplo de cómo usar ThreadPoolExecutor

Recuperando valor de cómputo - Callable

Si su cómputo produce algún valor de retorno que luego se requiere, una simple tarea Ejecutable no es suficiente. Para tales casos, puede usar ExecutorService.submit( Callable <T>) que devuelve un valor una vez que se completa la ejecución.

El Servicio devolverá un Future que puede utilizar para recuperar el resultado de la ejecución de la tarea.

// Submit a callable for execution
ExecutorService pool = anExecutorService;
Future<Integer> future = pool.submit(new Callable<Integer>() {
    @Override public Integer call() {
        //do some computation
        return new Random().nextInt();
    }
});    
// ... perform other tasks while future is executed in a different thread

Cuando necesite obtener el resultado del futuro, llame a future.get()

• Espera indefinidamente a que el futuro termine con un resultado.

    try {
        // Blocks current thread until future is completed
        Integer result = future.get(); 
    catch (InterruptedException || ExecutionException e) {
        // handle appropriately
    }
  

• Espere a que el futuro termine, pero no más de lo especificado.

    try {
        // Blocks current thread for a maximum of 500 milliseconds.
        // If the future finishes before that, result is returned,
        // otherwise TimeoutException is thrown.
        Integer result = future.get(500, TimeUnit.MILLISECONDS); 
    catch (InterruptedException || ExecutionException || TimeoutException e) {
        // handle appropriately
    }
  

Si ya no es necesario el resultado de una tarea programada o en ejecución, puede llamar a Future.cancel(boolean) para cancelarla.

• Llamar a cancel(false) solo eliminará la tarea de la cola de tareas que se ejecutarán.
• Llamar a cancel(true) también interrumpirá la tarea si se está ejecutando actualmente.

Programar tareas para que se ejecuten a una hora determinada, después de un retraso o repetidamente

La clase ScheduledExecutorService proporciona métodos para programar tareas únicas o repetidas de varias maneras. El siguiente ejemplo de código supone que el pool se ha declarado e inicializado de la siguiente manera:

ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);

Además de los métodos normales de ExecutorService , la API ScheduledExecutorService agrega 4 métodos que programan tareas y devuelven objetos ScheduledFuture . Este último se puede utilizar para recuperar resultados (en algunos casos) y cancelar tareas.

Iniciar una tarea después de un retraso fijo

El siguiente ejemplo programa una tarea para comenzar después de diez minutos.

ScheduledFuture<Integer> future = pool.schedule(new Callable<>() {
        @Override public Integer call() {
            // do something
            return 42;
        }
    }, 
    10, TimeUnit.MINUTES);

Comenzando tareas a una tasa fija

El siguiente ejemplo programa una tarea para comenzar después de diez minutos, y luego repetidamente a una velocidad de una vez cada minuto.

ScheduledFuture<?> future = pool.scheduleAtFixedRate(new Runnable() {
        @Override public void run() {
            // do something
        }
    }, 
    10, 1, TimeUnit.MINUTES);

La ejecución de la tarea continuará de acuerdo con el cronograma hasta que la pool se cierre, el future se cancele o una de las tareas encuentre una excepción.

Se garantiza que las tareas programadas por una llamada scheduledAtFixedRate AtFixedRate no se superpondrán en el tiempo. Si una tarea lleva más tiempo que el período prescrito, las ejecuciones de la siguiente tarea y las posteriores pueden comenzar tarde.

Comenzando tareas con un retraso fijo

El siguiente ejemplo programa una tarea para comenzar después de diez minutos, y luego repetidamente con un retraso de un minuto entre el final de una tarea y el inicio de la siguiente.

ScheduledFuture<?> future = pool.scheduleWithFixedDelay(new Runnable() {
        @Override public void run() {
            // do something
        }
    }, 
    10, 1, TimeUnit.MINUTES);

La ejecución de la tarea continuará de acuerdo con el cronograma hasta que la pool se cierre, el future se cancele o una de las tareas encuentre una excepción.

Manejar Ejecución Rechazada

Si

1. intenta enviar tareas a un Ejecutor de apagado o
2. la cola está saturada (solo es posible con las encuadernadas) y se ha alcanzado el número máximo de subprocesos,

RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor) .

El comportamiento predeterminado es que recibirá una excepción RejectedExecutionException en la persona que llama. Pero hay más comportamientos predefinidos disponibles:

• ThreadPoolExecutor.AbortPolicy (por defecto, lanzará REE)
• ThreadPoolExecutor.CallerRunsPolicy (ejecuta la tarea en el subproceso de la persona que llama, bloqueándolo )
• ThreadPoolExecutor.DiscardPolicy (descartar tarea silenciosamente)
• ThreadPoolExecutor.DiscardOldestPolicy (descarta silenciosamente la tarea más antigua en la cola y reintenta la ejecución de la nueva tarea)

Puedes establecerlos usando uno de los constructores de ThreadPool:

public ThreadPoolExecutor(int corePoolSize,
                      int maximumPoolSize,
                      long keepAliveTime,
                      TimeUnit unit,
                      BlockingQueue<Runnable> workQueue,
                      RejectedExecutionHandler handler) // <--

public ThreadPoolExecutor(int corePoolSize,
                      int maximumPoolSize,
                      long keepAliveTime,
                      TimeUnit unit,
                      BlockingQueue<Runnable> workQueue,
                      ThreadFactory threadFactory,
                      RejectedExecutionHandler handler) // <--

También puede implementar su propio comportamiento al extender la interfaz RejectedExecutionHandler :

void rejectedExecution(Runnable r, ThreadPoolExecutor executor)

diferencias de manejo de excepciones de submit () vs execute ()

Generalmente, el comando de ejecución () se usa para las llamadas de disparo y olvido (sin necesidad de analizar el resultado) y el comando de envío () se usa para analizar el resultado del objeto futuro.

Debemos ser conscientes de la diferencia clave de los mecanismos de manejo de excepciones entre estos dos comandos.

El marco se traga las excepciones de submit () si no las atrapó.

Código de ejemplo para entender la diferencia:

Caso 1: envíe el comando Ejecutable con ejecutar (), que informa la Excepción.

import java.util.concurrent.*;
import java.util.*;

public class ExecuteSubmitDemo {
    public ExecuteSubmitDemo() {
        System.out.println("creating service");
        ExecutorService service = Executors.newFixedThreadPool(2);
        //ExtendedExecutor service = new ExtendedExecutor();
        for (int i = 0; i < 2; i++){
            service.execute(new Runnable(){
                 public void run(){
                    int a = 4, b = 0;
                    System.out.println("a and b=" + a + ":" + b);
                    System.out.println("a/b:" + (a / b));
                    System.out.println("Thread Name in Runnable after divide by zero:"+Thread.currentThread().getName());
                 }
            });
        }
        service.shutdown();
    }
    public static void main(String args[]){
        ExecuteSubmitDemo demo = new ExecuteSubmitDemo();
    }
}

class ExtendedExecutor extends ThreadPoolExecutor {

   public ExtendedExecutor() { 
       super(1, 1, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100));
   }
   // ...
   protected void afterExecute(Runnable r, Throwable t) {
     super.afterExecute(r, t);
     if (t == null && r instanceof Future<?>) {
       try {
         Object result = ((Future<?>) r).get();
       } catch (CancellationException ce) {
           t = ce;
       } catch (ExecutionException ee) {
           t = ee.getCause();
       } catch (InterruptedException ie) {
           Thread.currentThread().interrupt(); // ignore/reset
       }
     }
     if (t != null)
       System.out.println(t);
   }
 }

salida:

creating service
a and b=4:0
a and b=4:0
Exception in thread "pool-1-thread-1" Exception in thread "pool-1-thread-2" java.lang.ArithmeticException: / by zero
        at ExecuteSubmitDemo$1.run(ExecuteSubmitDemo.java:15)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:744)
java.lang.ArithmeticException: / by zero
        at ExecuteSubmitDemo$1.run(ExecuteSubmitDemo.java:15)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:744)

Caso 2: Reemplace execute () con submit (): service.submit(new Runnable(){ En este caso, Framework se traga las excepciones ya que el método run () no las detectó explícitamente.

salida:

creating service
a and b=4:0
a and b=4:0

Caso 3: Cambie el nuevoFixedThreadPool a ExtendedExecutor

//ExecutorService service = Executors.newFixedThreadPool(2);
 ExtendedExecutor service = new ExtendedExecutor(); 

salida:

creating service
a and b=4:0
java.lang.ArithmeticException: / by zero
a and b=4:0
java.lang.ArithmeticException: / by zero

He demostrado este ejemplo para cubrir dos temas: Use su ThreadPoolExecutor personalizado y maneje Exectpion con ThreadPoolExecutor personalizado.

Otra solución simple al problema anterior: cuando está utilizando el comando ExecutorService & submit normal, obtenga el objeto Future del comando submit () call call get () en Future. Capte las tres excepciones, que se han citado en la implementación del método afterExecute. Ventaja de ThreadPoolExecutor personalizado sobre este enfoque: tiene que manejar el mecanismo de manejo de excepciones en un solo lugar: ThreadPoolExecutor personalizado.

Casos de uso para diferentes tipos de construcciones de concurrencia.

1. EjecutorServicio

   ExecutorService executor = Executors.newFixedThreadPool(50);

   Es simple y fácil de usar. Oculta los detalles de bajo nivel de ThreadPoolExecutor .

   Prefiero este cuando el número de tareas Callable/Runnable es pequeño en número y la acumulación de tareas en una cola ilimitada no aumenta la memoria y degrada el rendimiento del sistema. Si tiene restricciones de CPU/Memory , prefiero usar ThreadPoolExecutor con restricciones de capacidad y RejectedExecutionHandler para manejar el rechazo de tareas.

2. CountDownLatch

   CountDownLatch se inicializará con un recuento determinado. Este conteo es disminuido por las llamadas al método countDown() . Los hilos que esperan que este conteo llegue a cero pueden llamar a uno de los métodos await() . La llamada await() bloquea el hilo hasta que el conteo llega a cero. Esta clase permite que un subproceso java espere hasta que otro conjunto de subprocesos complete sus tareas.

   Casos de uso:

   1. Lograr el máximo paralelismo: a veces queremos iniciar una serie de subprocesos al mismo tiempo para lograr el máximo paralelismo

   2. Espere N hilos para completar antes de iniciar la ejecución

   3. Detección de interbloqueo.

3. ThreadPoolExecutor : Proporciona más control. Si la aplicación está restringida por la cantidad de tareas pendientes ejecutables / recuperables, puede usar la cola delimitada configurando la capacidad máxima. Una vez que la cola alcanza su capacidad máxima, puede definir RejectionHandler. Java proporciona cuatro tipos de políticas RejectedExecutionHandler .

   1. ThreadPoolExecutor.AbortPolicy , el controlador lanza una ThreadPoolExecutor.AbortPolicy RejectedExecutionException en tiempo de ejecución al rechazarla.

   2. ThreadPoolExecutor.CallerRunsPolicy`, el hilo que invoca a sí mismo ejecuta la tarea. Esto proporciona un mecanismo de control de retroalimentación simple que reducirá la velocidad a la que se envían las nuevas tareas.

   3. En ThreadPoolExecutor.DiscardPolicy , una tarea que no se puede ejecutar simplemente se elimina.

   4. ThreadPoolExecutor.DiscardOldestPolicy , si el ejecutor no se apaga, la tarea en la cabecera de la cola de trabajo se descarta y, a continuación, se vuelve a intentar la ejecución (lo que puede fallar de nuevo, haciendo que esto se repita).

Si desea simular CountDownLatch comportamiento, puede utilizar invokeAll() método.

4. Un mecanismo más que no ha citado es ForkJoinPool

   El ForkJoinPool se agregó a Java en Java 7. El ForkJoinPool es similar al Java ExecutorService pero con una diferencia. ForkJoinPool facilita que las tareas dividan su trabajo en tareas más pequeñas que luego se envían también a ForkJoinPool . El robo de tareas ocurre en ForkJoinPool cuando los subprocesos de trabajo libres roban tareas de la cola de subprocesos de trabajo ocupado.

   Java 8 ha introducido una API más en ExecutorService para crear un grupo de robo de trabajo. No tiene que crear RecursiveTask y RecursiveAction pero aún puede usar ForkJoinPool .

   public static ExecutorService newWorkStealingPool()
   

   Crea un grupo de subprocesos de robo de trabajo utilizando todos los procesadores disponibles como su nivel de paralelismo objetivo.

   Por defecto, tomará el número de núcleos de CPU como parámetro.

Todos estos cuatro mecanismos son complementarios entre sí. Dependiendo del nivel de granularidad que desee controlar, debe elegir los correctos.

Espere a que se completen todas las tareas en ExecutorService

Echemos un vistazo a varias opciones para esperar a que se completen las tareas enviadas a Executor

1. ExecutorService invokeAll()

   Ejecuta las tareas dadas, devolviendo una lista de futuros que mantienen su estado y resultados cuando todo está completado.

Ejemplo:

import java.util.concurrent.*;
import java.util.*;

public class InvokeAllDemo{
    public InvokeAllDemo(){
        System.out.println("creating service");
        ExecutorService service = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        
        List<MyCallable> futureList = new ArrayList<MyCallable>();
        for (int i = 0; i < 10; i++){
            MyCallable myCallable = new MyCallable((long)i);
            futureList.add(myCallable);
        }
        System.out.println("Start");
        try{
            List<Future<Long>> futures = service.invokeAll(futureList);  
        } catch(Exception err){
            err.printStackTrace();
        }
        System.out.println("Completed");
        service.shutdown();
    }
    public static void main(String args[]){
        InvokeAllDemo demo = new InvokeAllDemo();
    }
    class MyCallable implements Callable<Long>{
        Long id = 0L;
        public MyCallable(Long val){
            this.id = val;
        }
        public Long call(){
            // Add your business logic
            return id;
        }
    }
}

2. CountDownLatch

   Una ayuda de sincronización que permite que uno o más subprocesos esperen hasta que se complete un conjunto de operaciones que se están realizando en otros subprocesos.

   Un CountDownLatch se inicializa con un recuento dado. El bloque de métodos de espera hasta que el conteo actual llegue a cero debido a invocaciones del método countDown() , después de lo cual se liberan todos los subprocesos en espera y cualquier invocación posterior de espera regresa de inmediato. Este es un fenómeno de un solo disparo: el conteo no se puede reiniciar. Si necesita una versión que restablezca la cuenta, considere usar un CyclicBarrier .

3. ForkJoinPool o newWorkStealingPool() en Ejecutores

4. Iterar a través de todos los objetos Future creados después de enviar a ExecutorService

5. Forma recomendada de apagar desde la página de documentación de Oracle de ExecutorService :

   void shutdownAndAwaitTermination(ExecutorService pool) {
       pool.shutdown(); // Disable new tasks from being submitted
       try {
         // Wait a while for existing tasks to terminate
         if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
           pool.shutdownNow(); // Cancel currently executing tasks
           // Wait a while for tasks to respond to being cancelled
           if (!pool.awaitTermination(60, TimeUnit.SECONDS))
               System.err.println("Pool did not terminate");
         }
       } catch (InterruptedException ie) {
         // (Re-)Cancel if current thread also interrupted
         pool.shutdownNow();
         // Preserve interrupt status
         Thread.currentThread().interrupt();
       }
   

   shutdown(): inicia un cierre ordenado en el que se ejecutan las tareas enviadas anteriormente, pero no se aceptarán nuevas tareas.

   shutdownNow(): intenta detener todas las tareas en ejecución activa, detiene el procesamiento de las tareas en espera y devuelve una lista de las tareas que estaban pendientes de ejecución.

   En el ejemplo anterior, si sus tareas tardan más tiempo en completarse, puede cambiar si la condición se convierte en condición condicional.

   Reemplazar

   if (!pool.awaitTermination(60, TimeUnit.SECONDS))
   

   con

   while(!pool.awaitTermination(60, TimeUnit.SECONDS)) {
     Thread.sleep(60000);
   

   }

Casos de uso para diferentes tipos de servicios de ejecución

Los ejecutores devuelven diferentes tipos de ThreadPools para satisfacer necesidades específicas.

1. public static ExecutorService newSingleThreadExecutor()

   Crea un Ejecutor que utiliza un único subproceso de trabajo que opera en una cola ilimitada

   Hay una diferencia entre newFixedThreadPool(1) y newSingleThreadExecutor() como dice el documento java para este último:

   A diferencia del newFixedThreadPool (1) equivalente, se garantiza que el ejecutor devuelto no se puede volver a configurar para utilizar subprocesos adicionales.

   Lo que significa que un newFixedThreadPool se puede reconfigurar más adelante en el programa por: ((ThreadPoolExecutor) fixedThreadPool).setMaximumPoolSize(10) Esto no es posible para newSingleThreadExecutor

   Casos de uso:

   1. Desea ejecutar las tareas enviadas en una secuencia.
   2. Solo necesitas un hilo para manejar toda tu solicitud

   Contras:

   1. La cola no enlazada es dañina

2. public static ExecutorService newFixedThreadPool(int nThreads)

   Crea un grupo de subprocesos que reutiliza un número fijo de subprocesos que operan en una cola compartida ilimitada. En cualquier momento, como máximo, las hebras nThreads serán tareas de procesamiento activas. Si se envían tareas adicionales cuando todos los subprocesos están activos, esperarán en la cola hasta que haya un subproceso disponible

   Casos de uso:

   1. Uso efectivo de los núcleos disponibles. Configure nThreads como Runtime.getRuntime().availableProcessors() nThreads Runtime.getRuntime().availableProcessors()
   2. Cuando decide que el número de subprocesos no debe exceder un número en el grupo de subprocesos

   Contras:

   1. La cola no unida es dañina.

3. public static ExecutorService newCachedThreadPool()

   Crea un grupo de subprocesos que crea nuevos subprocesos según sea necesario, pero reutilizará los subprocesos construidos previamente cuando estén disponibles.

   Casos de uso:

   1. Para tareas asíncronas de corta duración.

   Contras:

   1. La cola no unida es dañina.
   2. Cada nueva tarea creará un nuevo hilo si todos los hilos existentes están ocupados. Si la tarea está tomando una larga duración, se crearán más hilos, lo que degradará el rendimiento del sistema. Alternativa en este caso: newFixedThreadPool

4. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

   Crea un grupo de subprocesos que puede programar comandos para que se ejecuten después de un retraso determinado o para que se ejecuten periódicamente.

   Casos de uso:

   1. Manejo de eventos recurrentes con retrasos, lo que ocurrirá en el futuro en cierto intervalo de tiempo

   Contras:

   1. La cola no unida es dañina.

   5. public static ExecutorService newWorkStealingPool()

   Crea un grupo de subprocesos de robo de trabajo utilizando todos los procesadores disponibles como su nivel de paralelismo objetivo

   Casos de uso:

   1. Para dividir y vencer tipo de problemas.
   2. Uso efectivo de hilos inactivos. Subprocesos inactivos roba tareas de subprocesos ocupados.

   Contras:

   1. El tamaño de la cola sin límite es perjudicial.

Puede ver un inconveniente común en todos estos ExecutorService: cola ilimitada. Esto será abordado con ThreadPoolExecutor

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, 
TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler)

Con ThreadPoolExecutor , puedes

1. Controlar el tamaño de la agrupación de hilos dinámicamente
2. Establecer la capacidad para BlockingQueue
3. Define RejectionExecutionHander cuando la cola está llena
4. CustomThreadFactory para agregar alguna funcionalidad adicional durante la creación de Thread (public Thread newThread(Runnable r)

Uso de grupos de subprocesos

Grupos de subprocesos se utilizan principalmente métodos de llamada en ExecutorService .

Se pueden usar los siguientes métodos para enviar trabajo para ejecución:

Una vez que haya terminado con el grupo de subprocesos, puede llamar a shutdown() para terminar el grupo de subprocesos. Esto ejecuta todas las tareas pendientes. Para esperar a que se ejecuten todas las tareas, puede recorrer awaitTermination o isShutdown() .