Ejecutores de Java: ¿cómo recibir una notificación, sin bloqueo, cuando se completa una tarea?

StackOverflow https://stackoverflow.com/questions/826212

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:

  1. Tomar una tarea de la cola
  2. Enviarlo al albacea
  3. Llame .get en el futuro devuelto y bloquee hasta que haya un resultado disponible
  4. 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)

¿Fue útil?

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) y afterExecute (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, reinicializando ThreadLocals , recopilando estadísticas o agregando entradas de registro. Además, el método terminated () puede anularse para realizar cualquier procesamiento especial que deba realizarse una vez que el Executor 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:

  1. Si desea procesar tareas en secuencia en orden FIFO, reemplace newFixedThreadPool (5) con newFixedThreadPool (1)

  2. 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());

  3. Puede reemplazar newFixedThreadPool () con uno de 

    Executors.newCachedThreadPool()
Executors.newWorkStealingPool()
ThreadPoolExecutor

    dependiendo de su caso de uso.

  4. 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 invocable

    b. Convierta su método Callable en tarea Callable / 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);
Licenciado bajo: CC-BY-SA con atribución
No afiliado a StackOverflow
scroll top