Executor Framework
Cuando un programa ejecuta muchas tareas concurrentes, todo el código
relacionado con los threads tiene que ser implementado, se tiene
que crear un objeto por thread, ejecutar el thread, obtener sus
resultados, y así sucesivamente. Esto puede traer algunos problemas
tales como no administrar de manera eficiente los recursos de la
máquina y afectar el rendimiento de la aplicación. Para aplicaciones de gran tamaño se necesita un mejor enfoque y el
executor freamwork puede ayudar con esto.
El executor freamwork separa la tarea de creación del thread, la
ejecución y su administración, encapsula la funcionalidad y mejora
el rendimiento usando un “pool” de threads
La forma en que el executor freamwor trabaja es muy simple, sólo
requiere de instancias de objetos de tipo Runnable o Callable y el se
encarga del resto.
Ejemplo:
public class Servidor {
private ThreadPoolExecutor executor;
public Servidor() {
executor = (ThreadPoolExecutor)Executors.newCachedThreadPool();//Crea el objeto executor o un pool de threads
}
public void ejecutaTarea(Task tarea) {
executor.execute(tarea);// Ejecuta una tarea
}
public void terminaServidor () {
executor.shutdown();// Este metodo termina el executor
}
}
public class Tarea implements Runnable {
private String nombre;
public Task(String nombre) {
this.nombre = nombre;
}
public void run() {
System.out.println(Thread.currentThread().getName() + ", creado: " + new Date());
try {
TimeUnit.SECONDS.sleep((long) (Math.random() * 10));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + ", terminado: " + new Date());
}
}
...........
Servidor servidor = new Servidor();
for (int i = 0; i < 10; i++) {
Tarea tarea = new Tarea("Tarea " + i);
servidor.ejecutaTarea(tarea);
}
servidor.endServer();
...........
En el ejemplo anterior, la clase Executors creo un objeto de tipo
java.util.concurrent.ThreadPoolExecutor, esta clase es una
implementación de la interfaz java.util.concurrent.ExecutorService.
Aunque ThreadPoolExecutor puede ser creado usando directamente sus
constructores, se recomienda utilizar la clase Executors.
El ThreadPoolExecutor utiliza el método execute() para ejecutar un
Runnable o Callable. Tambien tiene otros métodos como getPoolSize(),
getCompleteTaskCount () para obtener el estado del pool.
El ThreadPoolExecutor tiene que ser terminada explícitamente
llamando al método endServer(), de lo contrario no termina y el
programa nunca terminará.
Para evitar saturar la aplicación y provocar un mal rendimiento, la
clase Executors tiene el método newFixedThreadPool(int nthreads)
que crea un executor de threads de tamaño fijo. Este executor tiene un número máximo de threads indicados por el
parametrio nthreads, y como dice el java api "En cualquier
momento, como máximo un numero “nthreads ” de threads estarán
activos ejecutando tareas. Si tareas adicionales son enviadas cuando
todos los threads estén activos, van a tener que esperar en la cola
hasta que un thread este disponible. Si cualquier thread termina
debido a un error durante la ejecución antes de terminar el
executor, uno nuevo tomará su lugar si es necesario para ejecutar
tareas posteriores ".
Ejemplo:
......
public Servidor() {
executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(5);
}
......
En el ejemplo se crea un executor con un máximo de 5 threads, esto
significa que si más de 5 tareas se envían a ejecutar sólo 5 se
ejecutaran y el resto se bloqueará hasta que haya un thread libre
para procesarlos.
Tareas que
regresan un valor
El executor framework puede ejecutar tareas que devuelven un valor,
esta es otra ventaja de utilizar este framework. Para este mecanismo
se utiliza la interfaz java.util.concurrent.Callable, en lugar
de tener un método run(), se hace una llamada al call() que devuelve
cualquier tipo de objeto que es especificado de forma genérica:
public interface Callable<V> {
V call() throws Exception;
}
El ExecutorService tiene el método submit() que acepta
objetos de tipo Callable y los ejecuta, este método devuelve un
objeto del tipo java.util.concurrent.Future, la interfaz
Future tiene métodos para obtener el resultado generado por el
objeto Callable.
Ejemplo:
public class CalculadorMultiplicacion implements Callable<Integer> {
private int operador1;
private int operador2;
public CalculadorMultiplicacion(int operador1, int operador2) {
this.operador1 = operador1;
this.operador2 = operador2;
}
public Integer call() throws Exception {
return operador1 * operador2;
}
}
.......................
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);//Maximo 2 threads
List<Future<Integer>> listaResultados = new ArrayList<Future<Integer>>();
for (int i = 0; i < 10; i++) {
CalculadorMultiplicacion calculador = new CalculadorMultiplicacion((int)(Math.random()*10), (int)(Math.random()*10));
Future<Integer> resultado = executor.submit(calculador);
listaResultados.add(resultado);
}
while (executor.getCompletedTaskCount() < listaResultados.size()) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}//Espera a que las tareas se completen
for (int i = 0; i < listaResultados.size(); i++) {
Future<Integer> resultado = listaResultados.get(i);
try {
System.out.println("El resultado de la tarea "+i+ " es:" + resultado.get());
} catch (Exception e) {
e.printStackTrace();
}
}
.......................
En el ejemplo anterior, los objetos Callable realizan una operación
de multiplicación se envían al executor utilizando el método de
submit(), el programa espera hasta que todas las tareas hayan acabado
verificando con el método del executor getCompletedTaskCount (), una
vez que terminen los resultados se obtienen con el método get() del
objeto Future.
El ExecutorService tiene el método invokeAny(tareas) que
recibe una colección de tareas luego las ejecuta y devuelve el
resultado de la primer tarea que termina sin
lanzar una excepción, las tareas que no han finalizado se cancelan.
Un ejemplo en el que se podría utilizar este método podría ser
para buscar servicios, una aplicación que quiere encontrar una
conexión de base de datos en servidores diferentes, la primera tarea
que encuentre disponible el servicio es el que se va a utilizar, las
demás tareas son ignoradas.
El ejemplo anterior implementado con el método invokeAny () quedaría
de la siguiente forma:
.................
ExecutorService executor = (ExecutorService) Executors.newCachedThreadPool();
List<CalculadorMultiplicacion> listaTareas= new ArrayList<CalculadorMultiplicacion>();
for (int i = 0; i < 10; i++) {
CalculadorMultiplicacion calculador = new CalculadorMultiplicacion((int)(Math.random()*10), (int)(Math.random()*10));
listaTareas.add(calculador);
}
try {
Integer resultado = executor.invokeAny(listaTareas);
System.out.println("El resultado de la primer tarea en terminar es:" + resultado);
} catch (Exception e) {e.printStackTrace();}
//Termina el Executor
executor.shutdown();
.............
El ExecutorService tiene otro mecanismo para ejecutar múltiples
tareas y procesar el resultado de todas las tareas, el método
invokeAll (tareas) recibe una colección de tareas, los ejecuta y
devuelve una lista de objetos de tipo Future.
Ejemplo:
ExecutorService executor = (ExecutorService) Executors.newCachedThreadPool();
List<CalculadorMultiplicacion> listaTareas = new ArrayList<CalculadorMultiplicacion>();
for (int i = 0; i < 10; i++) {
CalculadorMultiplicacion calculador = new CalculadorMultiplicacion((int)(Math.random()*10), (int)(Math.random()*10));
listaTareas.add(calculador);
}
List<Future<Integer>> listaResultados = null;
try {
listaResultados = executor.invokeAll(listaTareas);
} catch (Exception e) {e.printStackTrace();}
executor.shutdown();
for (int i = 0; i < listaResultados.size(); i++) {
Future<Integer> resultado = listaResultados.get(i);
try {
System.out.println("El resultado de la tarea "+i+ " es:" + resultado.get());
} catch (Exception e) {
e.printStackTrace();
}
}
En el ejemplo en lugar de enviar cada tarea para el executor con el
método submit(), todas las tareas se agrupan en una lista y se
enviar para ejecutar a través de el método invokeAll().
Tareas
programadas
La clase Executores puede crear un “pool” que programa las tareas
después de un cierto tiempo, o las ejecuta periódicamente. Este
“pool” implementa la interfaz
java.util.concurrent.ScheduledExecutorService.
Ejemplo:
ScheduledExecutorService executor=(ScheduledExecutorService)Executors.newScheduledThreadPool(1);
List<Future<Integer>> listaResultados = new ArrayList<Future<Integer>>();
for (int i=0; i<5; i++) {
CalculadorMultiplicacion calculador = new CalculadorMultiplicacion((int)(Math.random()*10), (int)(Math.random()*10));
Future resultado<Integer> = executor.schedule(calculacor,i+1 , TimeUnit.SECONDS);
listaResultados.add(resultado);
}
executor.shutdown();
//Espera la terminación del executor
try {
executor.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
..............
En el ejemplo se crea un “pool” programado con un tamaño de 1,
cada thread es programado utilizando el método schedule(),
este método recibe como parámetros la tarea a ejecutar, el período
de tiempo de espera antes de la ejecución y la unidad de tiempo .
El executor utiliza el método awaitTermination() que bloquea
hasta que todas las tareas se han completado o pase el tiempo de
espera .
Tareas
rechazadas
Si una tarea se enviá al executor entre la llamada al método
shutdown() y el final de su ejecución, la tarea se rechaza. El
ejecutor proporciona un mecanismo para manejar esto, sólo requiere
una instancia de un objeto que implementa la interfaz
java.util.concurrent.RejectedExecutionHandler.
Example:
public class RejectedTaskController implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("La tarea ha sido rechazada");
}
}
........
RejectedTaskController controller = new RejectedTaskController();
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
executor.setRejectedExecutionHandler(controller);
........
Cuando se rechaza una tarea el método rejecedExecution() de
la instancia de RejectedExecutionHandler es llamado.
Ir a parte 6
Ir a parte 6
