Paquete java.util.concurrent

El paquete de concurrencia de Java cubre la concurrencia, subprocesos múltiples y paralelismo en la plataforma Java. La concurrencia es la capacidad de ejecutar varios o múltiples programas o aplicaciones en paralelo. La columna vertebral de la concurrencia de Java son los subprocesos (un proceso ligero, que tiene sus propios archivos y pilas y puede acceder a los datos compartidos de otros subprocesos en el mismo proceso). El rendimiento y la interactividad del programa se pueden mejorar realizando tareas que consumen mucho tiempo de forma asíncrona o en paralelo. Java 5 agregó un nuevo paquete a la plataforma java ⇾ paquete java.util.concurrent. Este paquete tiene un conjunto de clases e interfaces que ayudan en el desarrollo de aplicaciones concurrentes ( multihilo ).) en Java. Antes de este paquete, uno necesita hacer las clases de utilidad de su necesidad por su cuenta. 

Principales componentes/utilidades del paquete concurrente

1. Ejecutor
2. EjecutorService
3. ScheduledExecutorService
4. Futuro
5. pestillo de cuenta regresiva
6. Barrera cíclica
7. Semáforo
8. ThreadFactory
9. Cola de bloqueo
10. DelayQueue
11. Cerrar
12. fáser     

Ahora analicemos algunas de las utilidades más útiles de este paquete que son las siguientes: 

A. Ejecutor 

Executor es un conjunto de interfaces que representa un objeto cuya implementación ejecuta tareas. Depende de la implementación si la tarea debe ejecutarse en un subproceso nuevo o en un subproceso actual. Por lo tanto, podemos desacoplar el flujo de ejecución de tareas del mecanismo de ejecución de tareas real, utilizando esta interfaz. El ejecutor no requiere que la ejecución de la tarea sea asíncrona. La más simple de todas es la interfaz ejecutable.

public interface Executor {
    void execute( Runnable command );
}

Para crear una instancia de ejecutor , necesitamos crear un invocador.

public class Invoker implements Executor {
   @Override
   public void execute(Runnable r) {
       r.run();
   }
}

Ahora, para la ejecución de la tarea , podemos usar este invocador.

public void execute() {
   Executor exe = new Invoker();
   exe.execute( () -> {
       // task to be performed
   });
}

Si el ejecutor no puede aceptar la tarea a ejecutar, lanzará una excepción de ejecución rechazada .

B.ExecutorService

ExecutorService es una interfaz y solo obliga a la implementación subyacente a implementar el método execute() . Extiende la interfaz Executor y agrega una serie de métodos que ejecutan hilos que devuelven un valor. Los métodos para cerrar el grupo de subprocesos, así como la capacidad de implementar el resultado de la ejecución de la tarea.

Necesitamos crear un objetivo Runnable para usar ExecutorService.

public class Task implements Runnable {
   @Override
   public void run() {

       // task details
   }
}

Ahora, podemos crear un objeto/instancia de esta clase y asignar la tarea. Necesitamos especificar el tamaño del grupo de subprocesos al crear una instancia.

// 20 is the thread pool size
ExecutorService exec = Executors.newFixedThreadPool(20);

Para la creación de una instancia de ExecutorService de subproceso único , podemos usar newSingleThreadExecuter(ThreadFactory threadfactory) para crear la instancia. Después de crear el ejecutor, podemos enviar la tarea.

public void execute() {
   executor.submit(new Task());
}

Además, podemos crear una instancia Runnable para el envío de tareas.

executor.submit(() -> {
   new Task();
});

A continuación se enumeran dos métodos de terminación listos para usar:

1. shutdown(): espera hasta que finaliza la ejecución de todas las tareas enviadas.
2. shutdownNow(): Finaliza inmediatamente todas las tareas en ejecución/pendientes.

Hay un método más que es awaitTermination() que bloquea forzosamente hasta que todas las tareas hayan completado la ejecución después de que se haya producido un evento de apagado o se haya agotado el tiempo de espera de ejecución, o hasta que se interrumpa el subproceso de ejecución.

try {
   exec.awaitTermination( 50l, TimeUnit.NANOSECONDS );
} catch (InterruptedException e) {
   e.printStackTrace();
}

C. ScheduledExecutorService

Es similar a ExecutorService. La diferencia es que esta interfaz puede realizar tareas periódicamente. Tanto la función Runnable como la Callable se utilizan para definir la tarea.

public void execute() {
   ScheduledExecutorService execServ
     = Executors.newSingleThreadScheduledExecutor();

   Future<String> future = executorService.schedule(() -> {
       // ..
       return "Hello world";
   }, 1, TimeUnit.SECONDS);

   ScheduledFuture<?> scheduledFuture = execServ.schedule(() -> {
       // ..
   }, 1, TimeUnit.SECONDS);

   executorService.shutdown();
}

ScheduledExecutorService también puede definir una tarea después de un retraso fijo.

executorService.scheduleAtFixedRate(() -> {
   // ..
}, 1, 20, TimeUnit.SECONDS);

executorService.scheduleWithFixedDelay(() -> {
   // ..
}, 1, 20, TimeUnit.SECONDS);

Aquí,

• ScheduleAtFixedRate (comando ejecutable, retardo inicial largo, período largo, unidad de unidad de tiempo): este método crea y ejecuta una acción periódica que se invoca primero después del retraso inicial y, posteriormente, con el período dado hasta que se cierra la instancia del servicio.
• ScheduleWithFixedDelay (comando ejecutable, long initialDelay, long delay, unidad TimeUnit): este método crea y ejecuta una acción periódica que se invoca primero después del retraso inicial proporcionado y repetidamente con el retraso dado entre la terminación de la ejecución y la invocación del el proximo.

D. Futuro

Representa el resultado de una operación asíncrona. Los métodos que contiene comprueban si la operación asincrónica se completó o no, obtienen el resultado completo, etc. La API cancel(boolean isInterruptRunning) cancela la operación y libera el subproceso en ejecución. Si el valor de isInterruptRunning es verdadero, el subproceso que ejecuta la tarea finalizará instantáneamente. De lo contrario, todas las tareas en curso se completan.

El fragmento de código crea una instancia de Future.

public void invoke() {
   ExecutorService executorService = Executors.newFixedThreadPool(20);

   Future<String> future = executorService.submit(() -> {
       // ...
       Thread.sleep(10000l);
       return "Hello";
   });
}

El código para verificar si el resultado del futuro está listo o no y obtiene los datos cuando se realiza el cálculo.

if (future.isDone() && !future.isCancelled()) {
   try {
       str = future.get();
   } catch (InterruptedException | ExecutionException e) {
       e.printStackTrace();
   }
}

Especificación de tiempo de espera para una operación dada. Si el tiempo empleado es mayor que este tiempo, se lanza TimeoutException .

try {
   future.get(20, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
   e.printStackTrace();
} 

E. Cierre de cuenta regresiva

Es una clase de utilidad que bloquea un conjunto de subprocesos hasta que se completan algunas operaciones. Un CountDownLatch se inicializa con un contador (que es de tipo Integer). Este contador disminuye a medida que se completa la ejecución de los subprocesos dependientes. Pero una vez que el contador llega a cero, se liberan otros subprocesos.

F. Barrera cíclica

CyclicBarrier es casi lo mismo que CountDownLatch excepto que podemos reutilizarlo. Permite que múltiples subprocesos se esperen entre sí usando await() antes de invocar la tarea final y esta característica no está en CountDownLatch.

Estamos obligados a crear una instancia de Runnable Task para iniciar la condición de barrera.

public class Task implements Runnable {

   private CyclicBarrier barrier;

   public Task(CyclicBarrier barrier) {
       this.barrier = barrier;
   }

   @Override
   public void run() {
       try {
           LOG.info(Thread.currentThread().getName() +
             " is waiting");
           barrier.await();
           LOG.info(Thread.currentThread().getName() +
             " is released");
       } catch (InterruptedException | BrokenBarrierException e) {
           e.printStackTrace();
       }
   }

}

Ahora, invocando algunos subprocesos para competir con la condición de barrera:

public void start() {

   CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
       // ..
       LOG.info("All previous tasks completed");
   });

   Thread t11 = new Thread(new Task(cyclicBarrier), "T11");
   Thread t12 = new Thread(new Task(cyclicBarrier), "T12");
   Thread t13 = new Thread(new Task(cyclicBarrier), "T13");

   if (!cyclicBarrier.isBroken()) {
       t11.start();
       t12.start();
       t13.start();
   }
}

En el código anterior, el método isBroken() verifica si alguno de los subprocesos se interrumpió durante el tiempo de ejecución.

G. Semáforo

Se utiliza para bloquear el acceso a nivel de hilo a alguna parte del recurso lógico o físico. Semaphore contiene un conjunto de permisos. Dondequiera que el subproceso intente ingresar la parte del código de una sección crítica, el semáforo otorga el permiso, ya sea que el permiso esté disponible o no, lo que significa que la sección crítica está disponible o no. Si el permiso no está disponible, entonces el subproceso no puede ingresar a la sección crítica.

Es básicamente una variable llamada contador que mantiene el conteo de hilos entrantes y salientes de la sección crítica. Cuando el subproceso de ejecución libera la sección crítica, el contador aumenta.

El siguiente código se utiliza para la implementación de Semaphore:

static Semaphore semaphore = new Semaphore(20);

public void execute() throws InterruptedException {

   LOG.info("Available : " + semaphore.availablePermits());
   LOG.info("No. of threads waiting to acquire: " +
     semaphore.getQueueLength());

   if (semaphore.tryAcquire()) {
       try {
           // 
       }
       finally {
           semaphore.release();
       }
   }

}

Los semáforos se pueden usar para implementar una estructura de datos similar a Mutex .

H. fábrica de hilos

Actúa como un grupo de subprocesos que crea un nuevo subproceso a pedido. ThreadFactory se puede definir como:

public class GFGThreadFactory implements ThreadFactory {
   private int threadId;
   private String name;

   public GFGThreadFactory(String name) {
       threadId = 1;
       this.name = name;
   }

   @Override
   public Thread newThread(Runnable r) {
       Thread t = new Thread(r, name + "-Thread_" + threadId);
       LOG.info("created new thread with id : " + threadId +
           " and name : " + t.getName());
       threadId++;
       return t;
   }
}

I. Cola de bloqueo

La interfaz BlockingQueue admite el control de flujo (además de la cola) al introducir el bloqueo si BlockingQueue está lleno o vacío. Un subproceso que intenta poner en cola un elemento en una cola completa se bloquea hasta que otro subproceso hace espacio en la cola, ya sea eliminando uno o más elementos o borrando la cola por completo. De manera similar, bloquea un subproceso que intenta eliminar de una cola vacía hasta que otros subprocesos insertan un elemento. BlockingQueue no acepta un valor nulo. Si intentamos poner en cola el elemento nulo, arroja NullPointerException .

J: cola de retraso

DelayQueue es una cola de prioridad especializada que ordena elementos en función de su tiempo de retraso. Significa que solo se pueden tomar de la cola aquellos elementos cuyo tiempo haya expirado. El encabezado DelayQueue contiene el elemento que ha expirado en el menor tiempo. Si no ha expirado ningún retraso, entonces no hay encabezado y la encuesta devolverá un valor nulo. DelayQueue acepta solo aquellos elementos que pertenecen a una clase de tipo Delayed. DelayQueue implementa el método getDelay() para devolver el tiempo de retraso restante.

K: bloqueo

Es una utilidad para bloquear el acceso de otros subprocesos a un determinado segmento de código. La diferencia entre Lock y un bloque sincronizado es que tenemos la operación lock() y unlock() de las API de bloqueo en métodos separados, mientras que un bloque sincronizado está completamente contenido en los métodos.

L: Fáser

Es más flexible que CountDownLatch y CyclicBarrier. Phaser se utiliza para actuar como una barrera reutilizable en la que el número dinámico de subprocesos debe esperar antes de que continúe la ejecución. Se pueden coordinar varias fases de ejecución reutilizando la instancia de un Phaser para cada fase del programa.