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

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. Toma una tarea de la cola
2. Presentarlo 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. (Al parecer, funcionaljava y jetlang 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)

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 tareas Runnable 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()));
  }

}

En Java 8 puedes 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 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: mapUsersToUserViewsy updateView.

Use la futura API 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 FutureTaskclase y anular el done()método, luego agregar el FutureTaskobjeto al ExecutorService, para que el done()método invoque cuando se FutureTaskcomplete de inmediato.

ThreadPoolExecutorTambién tiene beforeExecutey afterExecutemétodos de enlace que pueden anular y hacer uso de. Aquí está la descripción de ThreadPoolExecutorlos Javadocs de .

Métodos de gancho

Esta clase proporciona métodos reemplazables beforeExecute(java.lang.Thread, java.lang.Runnable)y protegidos afterExecute(java.lang.Runnable, java.lang.Throwable)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, reinicializar ThreadLocals, recopilar estadísticas o agregar entradas de registro. Además, terminated()se puede anular el método para realizar cualquier procesamiento especial que deba realizarse una vez que Executorhaya 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.

Usar una 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 CountDownLatchy 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 una rutina 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.

Código simple para implementar Callbackmecanismo usandoExecutorService

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) connewFixedThreadPool(1)

2. Si desea procesar la siguiente tarea después de analizar el resultado callbackde la tarea anterior, simplemente descomente debajo de la línea

   //System.out.println("future status:"+future.get()+":"+future.isDone());

3. Puedes 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 asincrónica

   a. Pase un compartidoExecutorService or ThreadPoolExecutor tarea a invocable

   si. Convierte tu Callablemétodo aCallable/Runnable tarea

   C. Empuje la tarea de devolución de llamada a ExecutorService or ThreadPoolExecutor

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

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);

Esta es una extensión de la respuesta de Pache usando la guayaba ListenableFuture .

En particular, los Futures.transform()retornos ListenableFuturepueden 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 / 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).