13 RejectedExecutionHandler¶
Controlar las tareas rechazadas por el ejecutor¶
Cuando deseamos finalizar la ejecución de un ejecutor usamos su método shutdown() o shutdownNow() para indicarle que debe concluir su ejecución. En ese momento el ejecutor deja de aceptar nuevas tareas.
Si enviamos una tarea al ejecutor después de haberle informado de que debe terminarse, es decir, después de llamar al método shutdown() o shutdownNow(), dicha tarea es rechazada y, por defecto, la llamada al método con el que se ha enviado la tarea lanzará una excepción RejectedExecutionException.
La clase ThreadPoolExecutor proporciona un mecanismo para personalizar la respuesta ante el hecho de que una tarea sea rechazada. Para ello, podemos proporcionar al ejecutor un objeto de una clase que implemente la interfaz RejectedExecutionHandler, ya sea a través del constructor o a través del método setRejectedExecutionHandler(rejectedExecutionHandler).
La interfaz RejectedExecutionHandler define un único método rejectedExecution(runnable, threadPoolExecutor), que recibe el objeto Runnable correspondiente a la tarea rechazada y el ejecutor.
Si le hemos proporcionado al ejecutor un objeto RejectedExecutionHandler, cuando el ejecutor rechace una tarea, no lanzará la excepción RejectedExecutionException, sino que llamará al método rejectedExecution(runnable, threadPoolExecutor), pasándole la tarea rechazada y él mismo como ejecutor.
Proyecto RejectedExecutionHandler¶
Este proyecto es similar al proyecto FixedThreadPool que estudiamos anteriormente, pero se usará un objeto ThreadFactory para que el ejecutor cree sus hilos y un objeto RejectedExecutionHandler para tratar las tareas que hayan sido rechazadas por haber sido enviadas después de que el el servidor se haya terminado.
class Main {
public static void main(String[] args) {
Server server = new Server();
for (int i = 0; i < 50; i++) {
Task task = new Task("Task " + i);
server.execute(task);
try {
// The less time you sleep the greater the thread pool size gets.
// Try to reduce the time sleeping and see what happens to thread pool size.
Thread.sleep(100);
} catch (InterruptedException e) {
return;
}
}
try {
// server.shutdown();
// Try shutdownNow instead and see what happens.
server.shutdownNow();
} catch (InterruptedException exception) {
return;
}
Task task = new Task("Task sent after shutdown");
server.execute(task);
}
}
import java.util.concurrent.*;
class Server {
static class ServerThreadFactory implements ThreadFactory {
int threadNumber = 1;
@Override
public Thread newThread(Runnable runnable) {
return new Thread(runnable, "Server thread " + threadNumber++);
}
}
private final ThreadPoolExecutor fixedThreadPool =
(ThreadPoolExecutor) Executors.newFixedThreadPool(5);
Server() {
fixedThreadPool.setThreadFactory(new ServerThreadFactory());
fixedThreadPool.setRejectedExecutionHandler((runnable, executor) ->
System.out.printf("Handler -> Task rejected: %s\n",
((Task) runnable).getName()));
}
void execute(Task task) {
fixedThreadPool.execute(task);
}
@SuppressWarnings("unused")
void shutdown() throws InterruptedException {
fixedThreadPool.shutdown();
if (fixedThreadPool.awaitTermination(5, TimeUnit.SECONDS)) {
System.out.printf("Server -> Terminated. Completed: %d\n",
fixedThreadPool.getCompletedTaskCount());
} else {
System.out.printf("Server -> Await termination timeout. Completed: %d\n",
fixedThreadPool.getCompletedTaskCount());
}
}
void shutdownNow() throws InterruptedException {
fixedThreadPool.shutdownNow();
if (fixedThreadPool.awaitTermination(5, TimeUnit.SECONDS)) {
System.out.printf("Server -> Terminated. Completed: %d\n",
fixedThreadPool.getCompletedTaskCount());
} else {
System.out.printf("Server -> Await termination timeout. Completed: %d\n",
fixedThreadPool.getCompletedTaskCount());
}
}
}
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
class Task implements Runnable {
private final String name;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
Task(String name) {
this.name = name;
}
String getName() {
return name;
}
@Override
public void run() {
System.out.printf("%s -> %s -> Started at: %s\n",
Thread.currentThread().getName(), name, dateTimeFormatter.format(LocalDateTime.now()));
try {
work();
} catch (InterruptedException e) {
System.out.printf("%s -> %s -> Interrupted at: %s\n",
Thread.currentThread().getName(), name, dateTimeFormatter.format(LocalDateTime.now()));
return;
}
System.out.printf("%s -> %s -> Finished at: %s\n",
Thread.currentThread().getName(), name, dateTimeFormatter.format(LocalDateTime.now()));
}
private void work() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(10));
}
}