Executores Java: como ser notificado, sem bloquear, quando uma tarefa é concluída?

Digamos que eu tenha uma fila cheia de tarefas que preciso enviar para um serviço executor. Eu quero que eles sejam processados ​​um de cada vez. A maneira mais simples de pensar é:

1. Tome uma tarefa da fila
2. Envie para o executor
3. Ligue para .get no futuro retornado e bloqueie até que um resultado esteja disponível
4. Pegue outra tarefa da fila ...

No entanto, estou tentando evitar o bloqueio completo. Se eu tiver 10.000 filas desse tipo, que precisam que suas tarefas sejam processadas uma de cada vez, o espaço da pilha ficará sem espaço, pois a maioria delas estará mantendo os threads bloqueados.

O que eu gostaria é enviar uma tarefa e fornecer um retorno de chamada que é chamado quando a tarefa estiver concluída. Usarei essa notificação de retorno como sinalizador para enviar a próxima tarefa. (o funcionaljava e o jetlang aparentemente usam algoritmos sem bloqueio, mas não consigo entender o código deles)

Como posso fazer isso usando o java.util.concurrent do JDK, além de escrever meu próprio serviço executor?

(a fila que me alimenta essas tarefas pode ser bloqueada, mas esse é um problema a ser resolvido posteriormente)

Defina uma interface de retorno de chamada para receber os parâmetros que você deseja transmitir na notificação de conclusão. Em seguida, invoque-o no final da tarefa.

Você pode até escrever um wrapper geral para tarefas Runnable e enviá-las para ExecutorService. Ou veja abaixo um mecanismo incorporado ao 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();
  }

}

Com CompletableFuture, o Java 8 incluiu um meio mais elaborado de compor pipelines onde os processos podem ser concluídos de forma assíncrona e condicional. Aqui está um exemplo completo, mas completo, de notificação.

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

}

No Java 8, você pode usar o CompletableFuture . Aqui está um exemplo que eu tive no meu código em que estou usando-o para buscar usuários do meu serviço de usuário, mapeá-los para meus objetos de exibição e atualizar minha exibição ou mostrar uma caixa de diálogo de erro (este é um aplicativo GUI):

    CompletableFuture.supplyAsync(
            userService::listUsers
    ).thenApply(
            this::mapUsersToUserViews
    ).thenAccept(
            this::updateView
    ).exceptionally(
            throwable -> { showErrorDialogFor(throwable); return null; }
    );

Ele é executado de forma assíncrona. Estou usando dois métodos particulares: mapUsersToUserViewse updateView.

Use a futura API escutável do Guava e adicione um retorno de chamada. Cf. a partir do site :

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

Você pode estender a FutureTaskclasse e substituir o done()método e, em seguida, adicionar o FutureTaskobjeto ao ExecutorService, para que o done()método seja chamado quando FutureTaskconcluído imediatamente.

ThreadPoolExecutortambém possui métodos hook beforeExecutee afterExecuteque você pode substituir e usar. Aqui está a descrição ThreadPoolExecutordos Javadocs de .

Métodos de gancho

Esta classe fornece métodos beforeExecute(java.lang.Thread, java.lang.Runnable)e substituíveis protegidos afterExecute(java.lang.Runnable, java.lang.Throwable)chamados antes e após a execução de cada tarefa. Eles podem ser usados ​​para manipular o ambiente de execução; por exemplo, reinicializando ThreadLocals, reunindo estatísticas ou adicionando entradas de log. Além disso, o método terminated()pode ser substituído para executar qualquer processamento especial que precise ser feito após o Executortérmino completo. Se os métodos de gancho ou de retorno de chamada lançarem exceções, os encadeamentos internos do trabalhador poderão falhar e terminar abruptamente.

Use a CountDownLatch.

É de java.util.concurrente é exatamente o caminho de aguardar vários threads para concluir a execução antes de continuar.

Para obter o efeito de retorno de chamada que você está cuidando, isso exige um pouco de trabalho extra adicional. Ou seja, lidar com isso sozinho em um thread separado que usa CountDownLatche espera, e depois avisa o que você precisa notificar. Não há suporte nativo para retorno de chamada ou algo semelhante a esse efeito.

EDIT: agora que entendi melhor sua pergunta, acho que você está chegando longe demais, desnecessariamente. Se você fizer um regular SingleThreadExecutor, execute todas as tarefas e ele fará as filas de forma nativa.

Se você deseja garantir que nenhuma tarefa seja executada ao mesmo tempo, use um SingleThreadedExecutor . As tarefas serão processadas na ordem em que são enviadas. Você nem precisa realizar as tarefas, basta enviá-las ao executivo.

Código simples 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
    }
}

resultado:

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 principais:

1. Se você deseja processar tarefas em sequência na ordem FIFO, substitua newFixedThreadPool(5) pornewFixedThreadPool(1)

2. Se você deseja processar a próxima tarefa depois de analisar o resultado callbackda tarefa anterior, basta desmarcar a linha abaixo

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

3. Você pode substituir newFixedThreadPool()por um dos

   Executors.newCachedThreadPool()
   Executors.newWorkStealingPool()
   ThreadPoolExecutor
   

   dependendo do seu caso de uso.

4. Se você deseja manipular o método de retorno de chamada de forma assíncrona

   uma. Passar uma ExecutorService or ThreadPoolExecutortarefa compartilhada para Callable

   b. Converta seu Callablemétodo em Callable/Runnabletarefa

   c. Envie a tarefa de retorno de chamada para ExecutorService or ThreadPoolExecutor

Apenas para adicionar à resposta de Matt, que ajudou, aqui está um exemplo mais detalhado para mostrar o uso de um retorno de chamada.

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

A saída é:

    getPrimeAsync(); prime=241
    onPrimeListener; p=241
    Adios mi amigito

Você pode usar uma implementação de Callable de forma 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";
    }
}

onde CallbackInterface é algo muito básico como

public interface CallbackInterface {
    public int doSomething(int a, int b);
}

e agora a classe principal ficará assim

ExecutorService ex = Executors.newFixedThreadPool(2);

MyAsyncCallable<String> mac = new MyAsyncCallable<String>((a, b) -> a + b);
ex.submit(mac);

Esta é uma extensão da resposta do Pache usando goiabas ListenableFuture.

Em particular, os Futures.transform()retornos ListenableFuturepodem ser usados ​​para encadear chamadas assíncronas. Futures.addCallback()retorna void, portanto, não pode ser usado para encadeamento, mas é bom para lidar com sucesso / falha em uma conclusão assí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: Além de encadear tarefas assíncronas, Futures.transform()também permite agendar cada tarefa em um executor separado (não mostrado neste exemplo).