Ejecutores de Java: ¿cómo recibir una notificación, sin bloqueo, cuando se completa una tarea?
Pregunta
Digamos que tengo una cola llena de tareas que necesito enviar a un servicio de ejecutor. Quiero que se procesen uno a la vez. La forma más simple que se me ocurre es:
- Tomar una tarea de la cola
- Enviarlo al albacea
- Llame .get en el futuro devuelto y bloquee hasta que haya un resultado disponible
- Toma otra tarea de la cola ...
Sin embargo, estoy tratando de evitar el bloqueo por completo. Si tengo 10,000 de esas colas, que necesitan que sus tareas se procesen una a la vez, me quedaré sin espacio en la pila porque la mayoría de ellas se quedarán con hilos bloqueados.
Lo que me gustaría es enviar una tarea y proporcionar una devolución de llamada que se llama cuando se completa la tarea. Usaré esa notificación de devolución de llamada como una bandera para enviar la siguiente tarea. (Functionjava y Jetlang aparentemente usan algoritmos sin bloqueo, pero no entiendo su código)
¿Cómo puedo hacer eso usando java.util.concurrent de JDK, sin escribir mi propio servicio de ejecutor?
(la cola que me alimenta estas tareas puede bloquearse, pero ese es un problema que se abordará más adelante)
Solución
Defina una interfaz de devolución de llamada para recibir los parámetros que desee transmitir en la notificación de finalización. Luego invoque al final de la tarea.
Incluso podría escribir un contenedor general para las tareas Ejecutables y enviarlas a ExecutorService
. O vea a continuación un mecanismo integrado en Java 8.
class CallbackTask implements Runnable {
private final Runnable task;
private final Callback callback;
CallbackTask(Runnable task, Callback callback) {
this.task = task;
this.callback = callback;
}
public void run() {
task.run();
callback.complete();
}
}
Con CompletableFuture
, Java 8 incluyó un medio más elaborado para componer tuberías donde los procesos se pueden completar de forma asincrónica y condicional. Aquí hay un ejemplo artificial pero completo de notificación.
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class GetTaskNotificationWithoutBlocking {
public static void main(String... argv) throws Exception {
ExampleService svc = new ExampleService();
GetTaskNotificationWithoutBlocking listener = new GetTaskNotificationWithoutBlocking();
CompletableFuture<String> f = CompletableFuture.supplyAsync(svc::work);
f.thenAccept(listener::notify);
System.out.println("Exiting main()");
}
void notify(String msg) {
System.out.println("Received message: " + msg);
}
}
class ExampleService {
String work() {
sleep(7000, TimeUnit.MILLISECONDS); /* Pretend to be busy... */
char[] str = new char[5];
ThreadLocalRandom current = ThreadLocalRandom.current();
for (int idx = 0; idx < str.length; ++idx)
str[idx] = (char) ('A' + current.nextInt(26));
String msg = new String(str);
System.out.println("Generated message: " + msg);
return msg;
}
public static void sleep(long average, TimeUnit unit) {
String name = Thread.currentThread().getName();
long timeout = Math.min(exponential(average), Math.multiplyExact(10, average));
System.out.printf("%s sleeping %d %s...%n", name, timeout, unit);
try {
unit.sleep(timeout);
System.out.println(name + " awoke.");
} catch (InterruptedException abort) {
Thread.currentThread().interrupt();
System.out.println(name + " interrupted.");
}
}
public static long exponential(long avg) {
return (long) (avg * -Math.log(1 - ThreadLocalRandom.current().nextDouble()));
}
}
Otros consejos
En Java 8 puede usar CompletableFuture . Aquí hay un ejemplo que tenía en mi código donde lo estoy usando para buscar usuarios de mi servicio de usuario, asignarlos a mis objetos de vista y luego actualizar mi vista o mostrar un cuadro de diálogo de error (esta es una aplicación GUI):
CompletableFuture.supplyAsync(
userService::listUsers
).thenApply(
this::mapUsersToUserViews
).thenAccept(
this::updateView
).exceptionally(
throwable -> { showErrorDialogFor(throwable); return null; }
);
Se ejecuta de forma asincrónica. Estoy usando dos métodos privados: mapUsersToUserViews
y updateView
.
Use API futura escuchable de Guava y agregue una devolución de llamada. Cf. del sitio web:
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
ListenableFuture<Explosion> explosion = service.submit(new Callable<Explosion>() {
public Explosion call() {
return pushBigRedButton();
}
});
Futures.addCallback(explosion, new FutureCallback<Explosion>() {
// we want this handler to run immediately after we push the big red button!
public void onSuccess(Explosion explosion) {
walkAwayFrom(explosion);
}
public void onFailure(Throwable thrown) {
battleArchNemesis(); // escaped the explosion!
}
});
Puede extender la clase FutureTask
y anular el método done ()
, luego agregar el objeto FutureTask
al ExecutorService
, por lo que el método done ()
invocará cuando FutureTask
se complete de inmediato.
ThreadPoolExecutor
también tiene beforeExecute
y afterExecute
métodos de enlace que puede anular y utilizar. Aquí está la descripción de ThreadPoolExecutor
's Javadocs .
Métodos de enganche
Esta clase proporciona protecciones reemplazables
beforeExecute (java.lang.Thread, java.lang.Runnable)
yafterExecute (java .lang.Runnable, java.lang.Throwable)
métodos que se llaman antes y después de la ejecución de cada tarea. Estos pueden usarse para manipular el entorno de ejecución; por ejemplo, reinicializandoThreadLocals
, recopilando estadísticas o agregando entradas de registro. Además, el métodoterminated ()
puede anularse para realizar cualquier procesamiento especial que deba realizarse una vez que elExecutor
haya finalizado por completo. Si los métodos de enganche o devolución de llamada arrojan excepciones, los subprocesos internos del trabajador pueden fallar y terminar abruptamente.
Utilice un < code> CountDownLatch .
Es de java.util.concurrent
y es exactamente la forma de esperar que varios hilos completen la ejecución antes de continuar.
Para lograr el efecto de devolución de llamada que está buscando, eso requiere un poco de trabajo adicional adicional. Es decir, manejar esto usted mismo en un hilo separado que usa el CountDownLatch
y lo espera, luego continúa notificando lo que sea que necesite notificar. No hay soporte nativo para la devolución de llamada, ni nada similar a ese efecto.
EDITAR: ahora que entiendo mejor tu pregunta, creo que estás llegando demasiado lejos, innecesariamente. Si toma un regular SingleThreadExecutor
, dele todas las tareas y hará la cola de forma nativa.
Si desea asegurarse de que no se ejecutarán tareas al mismo tiempo, utilice un SingleThreadedExecutor . Las tareas se procesarán en el orden en que se envían. Ni siquiera necesita retener las tareas, solo envíelas al ejecutivo.
Solo para agregar a la respuesta de Matt, que ayudó, aquí hay un ejemplo más detallado para mostrar el uso de una devolución de llamada.
private static Primes primes = new Primes();
public static void main(String[] args) throws InterruptedException {
getPrimeAsync((p) ->
System.out.println("onPrimeListener; p=" + p));
System.out.println("Adios mi amigito");
}
public interface OnPrimeListener {
void onPrime(int prime);
}
public static void getPrimeAsync(OnPrimeListener listener) {
CompletableFuture.supplyAsync(primes::getNextPrime)
.thenApply((prime) -> {
System.out.println("getPrimeAsync(); prime=" + prime);
if (listener != null) {
listener.onPrime(prime);
}
return prime;
});
}
El resultado es:
getPrimeAsync(); prime=241
onPrimeListener; p=241
Adios mi amigito
Código simple para implementar el mecanismo Callback
usando ExecutorService
import java.util.concurrent.*;
import java.util.*;
public class CallBackDemo{
public CallBackDemo(){
System.out.println("creating service");
ExecutorService service = Executors.newFixedThreadPool(5);
try{
for ( int i=0; i<5; i++){
Callback callback = new Callback(i+1);
MyCallable myCallable = new MyCallable((long)i+1,callback);
Future<Long> future = service.submit(myCallable);
//System.out.println("future status:"+future.get()+":"+future.isDone());
}
}catch(Exception err){
err.printStackTrace();
}
service.shutdown();
}
public static void main(String args[]){
CallBackDemo demo = new CallBackDemo();
}
}
class MyCallable implements Callable<Long>{
Long id = 0L;
Callback callback;
public MyCallable(Long val,Callback obj){
this.id = val;
this.callback = obj;
}
public Long call(){
//Add your business logic
System.out.println("Callable:"+id+":"+Thread.currentThread().getName());
callback.callbackMethod();
return id;
}
}
class Callback {
private int i;
public Callback(int i){
this.i = i;
}
public void callbackMethod(){
System.out.println("Call back:"+i);
// Add your business logic
}
}
salida:
creating service
Callable:1:pool-1-thread-1
Call back:1
Callable:3:pool-1-thread-3
Callable:2:pool-1-thread-2
Call back:2
Callable:5:pool-1-thread-5
Call back:5
Call back:3
Callable:4:pool-1-thread-4
Call back:4
Notas clave:
- Si desea procesar tareas en secuencia en orden FIFO, reemplace
newFixedThreadPool (5)
connewFixedThreadPool (1)
-
Si desea procesar la siguiente tarea después de analizar el resultado de
callback
de la tarea anterior, simplemente descomente debajo de la línea//System.out.println("future status:"+future.get()+":"+future.isDone());
-
Puede reemplazar
newFixedThreadPool ()
con uno deExecutors.newCachedThreadPool() Executors.newWorkStealingPool() ThreadPoolExecutor
dependiendo de su caso de uso.
-
Si desea manejar el método de devolución de llamada de forma asíncrona
a. Pase un
ExecutorService o ThreadPoolExecutor
compartido a la tarea invocableb. Convierta su método
Callable
en tareaCallable / Runnable
c. Empuje la tarea de devolución de llamada a
ExecutorService o ThreadPoolExecutor
Esta es una extensión de la respuesta de Pache usando ListenableFuture
de Guava.
En particular, Futures.transform ()
devuelve ListenableFuture
para que pueda usarse para encadenar llamadas asíncronas. Futures.addCallback ()
devuelve void
, por lo que no se puede usar para encadenar, pero es bueno para manejar el éxito / el fracaso en una finalización asíncrona.
// ListenableFuture1: Open Database
ListenableFuture<Database> database = service.submit(() -> openDatabase());
// ListenableFuture2: Query Database for Cursor rows
ListenableFuture<Cursor> cursor =
Futures.transform(database, database -> database.query(table, ...));
// ListenableFuture3: Convert Cursor rows to List<Foo>
ListenableFuture<List<Foo>> fooList =
Futures.transform(cursor, cursor -> cursorToFooList(cursor));
// Final Callback: Handle the success/errors when final future completes
Futures.addCallback(fooList, new FutureCallback<List<Foo>>() {
public void onSuccess(List<Foo> foos) {
doSomethingWith(foos);
}
public void onFailure(Throwable thrown) {
log.error(thrown);
}
});
NOTA: además de encadenar tareas asíncronas, Futures.transform ()
también le permite programar cada tarea en un ejecutor separado (no se muestra en este ejemplo).
Puede usar una implementación de Callable de modo que
public class MyAsyncCallable<V> implements Callable<V> {
CallbackInterface ci;
public MyAsyncCallable(CallbackInterface ci) {
this.ci = ci;
}
public V call() throws Exception {
System.out.println("Call of MyCallable invoked");
System.out.println("Result = " + this.ci.doSomething(10, 20));
return (V) "Good job";
}
}
donde CallbackInterface es algo muy básico como
public interface CallbackInterface {
public int doSomething(int a, int b);
}
y ahora la clase principal se verá así
ExecutorService ex = Executors.newFixedThreadPool(2);
MyAsyncCallable<String> mac = new MyAsyncCallable<String>((a, b) -> a + b);
ex.submit(mac);