Java-Executoren: Wie kann ich benachrichtigt werden, ohne zu blockieren, wenn eine Aufgabe abgeschlossen ist?

154

Angenommen, ich habe eine Warteschlange voller Aufgaben, die ich an einen Executor-Service senden muss. Ich möchte, dass sie einzeln verarbeitet werden. Der einfachste Weg, den ich mir vorstellen kann, ist:

  1. Nehmen Sie eine Aufgabe aus der Warteschlange
  2. Senden Sie es an den Testamentsvollstrecker
  3. Rufen Sie .get für die zurückgegebene Zukunft auf und blockieren Sie, bis ein Ergebnis verfügbar ist
  4. Nehmen Sie eine andere Aufgabe aus der Warteschlange ...

Ich versuche jedoch zu vermeiden, vollständig zu blockieren. Wenn ich 10.000 solcher Warteschlangen habe, deren Aufgaben einzeln verarbeitet werden müssen, geht mir der Stapelspeicher aus, da die meisten von ihnen an blockierten Threads festhalten.

Ich möchte eine Aufgabe einreichen und einen Rückruf bereitstellen, der aufgerufen wird, wenn die Aufgabe abgeschlossen ist. Ich werde diese Rückrufbenachrichtigung als Flag verwenden, um die nächste Aufgabe zu senden. (FunctionalJava und Jetlang verwenden anscheinend solche nicht blockierenden Algorithmen, aber ich kann ihren Code nicht verstehen.)

Wie kann ich das mit java.util.concurrent von JDK tun, ohne meinen eigenen Executor-Service zu schreiben?

(Die Warteschlange, die mich mit diesen Aufgaben versorgt, blockiert möglicherweise selbst, aber das ist ein Problem, das später behoben werden muss.)

Shahbaz
quelle

Antworten:

146

Definieren Sie eine Rückrufschnittstelle, um alle Parameter zu erhalten, die Sie in der Abschlussbenachrichtigung weitergeben möchten. Rufen Sie es dann am Ende der Aufgabe auf.

Sie können sogar einen allgemeinen Wrapper für ausführbare Aufgaben schreiben und diese an senden ExecutorService. Oder sehen Sie unten einen in Java 8 integrierten Mechanismus.

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

}

Mit CompletableFutureJava 8 wurde ein ausgefeilteres Mittel zum Erstellen von Pipelines bereitgestellt, bei dem Prozesse asynchron und bedingt abgeschlossen werden können. Hier ist ein erfundenes, aber vollständiges Beispiel für eine Benachrichtigung.

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

}
erickson
quelle
1
Drei Antworten im Handumdrehen! Ich mag die CallbackTask, eine so einfache und unkomplizierte Lösung. Rückblickend sieht es offensichtlich aus. Vielen Dank. In Bezug auf andere Kommentare zu SingleThreadedExecutor: Ich habe möglicherweise Tausende von Warteschlangen, die Tausende von Aufgaben haben können. Jeder von ihnen muss seine Aufgaben einzeln bearbeiten, aber verschiedene Warteschlangen können parallel arbeiten. Deshalb verwende ich einen einzelnen globalen Threadpool. Ich bin neu bei Testamentsvollstreckern. Bitte sagen Sie mir, wenn ich mich irre.
Shahbaz
5
Gutes Muster, ich würde jedoch Guavas hörbare zukünftige API verwenden , die eine sehr gute Implementierung bietet.
Pierre-Henri
übertrifft dies nicht den Zweck der Verwendung von Future?
Pflege
2
@Zelphir Es war eine CallbackSchnittstelle, die Sie deklarieren; nicht aus einer Bibliothek. Heute würde ich wahrscheinlich verwenden Sie einfach Runnable, Consumeroder BiConsumer, je nachdem , was ich brauche von der Aufgabe auf den Hörer zurücklaufen.
Erickson
1
@Bhargav Dies ist typisch für Rückrufe - eine externe Entität "ruft" die steuernde Entität zurück. Möchten Sie, dass der Thread, der die Aufgabe erstellt hat, blockiert wird, bis die Aufgabe abgeschlossen ist? Welchen Zweck hat es dann, die Aufgabe in einem zweiten Thread auszuführen? Wenn Sie zulassen, dass der Thread fortgesetzt wird, muss er wiederholt einen freigegebenen Status überprüfen (wahrscheinlich in einer Schleife, hängt jedoch von Ihrem Programm ab), bis er eine Aktualisierung (Boolesches Flag, neues Element in der Warteschlange usw.) durch true feststellt Rückruf wie in dieser Antwort beschrieben. Es kann dann einige zusätzliche Arbeiten ausführen.
erickson
52

In Java 8 können Sie CompletableFuture verwenden . Hier ist ein Beispiel in meinem Code, in dem ich Benutzer aus meinem Benutzerdienst abrufe, sie meinen Ansichtsobjekten zuordne und dann meine Ansicht aktualisiere oder einen Fehlerdialog anzeige (dies ist eine GUI-Anwendung):

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

Es wird asynchron ausgeführt. Ich verwende zwei private Methoden: mapUsersToUserViewsund updateView.

Matt
quelle
Wie würde man eine CompletableFuture mit einem Executor verwenden? (um die Anzahl der gleichzeitigen / parallelen Instanzen zu begrenzen) Wäre dies ein Hinweis: Vgl.: Senden zukünftiger Aufgaben an einen Executor, warum funktioniert das ?
user1767316
47

Verwenden Sie die abhörbare zukünftige API von Guava und fügen Sie einen Rückruf hinzu. Vgl. von der Website:

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!
  }
});
Pierre-Henri
quelle
Hallo, aber wenn ich nach onSuccess diesen Thread stoppen möchte, wie kann ich das tun?
DevOps85
24

Sie können die FutureTaskKlasse erweitern, die done()Methode überschreiben und dann das FutureTaskObjekt zum hinzufügen ExecutorService, sodass die done()Methode FutureTasksofort nach Abschluss der Methode aufgerufen wird .

Auguste
quelle
then add the FutureTask object to the ExecutorServiceKönnten Sie mir bitte sagen, wie das geht?
Gary Gauh
@GaryGauh Weitere Informationen finden Sie unter FutureTask. Wir können es als MyFutureTask bezeichnen. Verwenden Sie dann ExcutorService, um MyFutureTask zu senden. Anschließend wird die Ausführungsmethode von MyFutureTask ausgeführt. Wenn MyFutureTask abgeschlossen ist, wird Ihre fertige Methode aufgerufen. Hier sind zwei FutureTask verwirrend, und tatsächlich ist MyFutureTask eine normale ausführbare Methode.
Chaojun Zhong
15

ThreadPoolExecutorhat auch beforeExecuteund afterExecuteHook-Methoden, die Sie überschreiben und verwenden können. Hier ist die Beschreibung von ThreadPoolExecutor's Javadocs .

Hook-Methoden

Diese Klasse bietet geschützte überschreibbare Methoden beforeExecute(java.lang.Thread, java.lang.Runnable)und afterExecute(java.lang.Runnable, java.lang.Throwable)Methoden, die vor und nach der Ausführung jeder Aufgabe aufgerufen werden. Diese können verwendet werden, um die Ausführungsumgebung zu manipulieren. Zum Beispiel Neuinitialisierung ThreadLocals, Sammeln von Statistiken oder Hinzufügen von Protokolleinträgen. Darüber hinaus kann die Methode terminated()überschrieben werden, um eine spezielle Verarbeitung durchzuführen, die ausgeführt werden muss, sobald die Executorvollständig beendet ist. Wenn Hook- oder Callback-Methoden Ausnahmen auslösen, können interne Worker-Threads fehlschlagen und abrupt beendet werden.

Cem Catikkas
quelle
6

Verwenden Sie a CountDownLatch.

Es ist von java.util.concurrentund es ist genau der Weg, um zu warten, bis mehrere Threads die Ausführung abgeschlossen haben, bevor Sie fortfahren.

Um den gewünschten Rückrufeffekt zu erzielen, ist ein wenig zusätzliche Arbeit erforderlich. Wenn Sie dies in einem separaten Thread selbst erledigen, der das verwendet CountDownLatchund darauf wartet, werden Sie dann benachrichtigt, was auch immer Sie benachrichtigen müssen. Es gibt keine native Unterstützung für Rückrufe oder ähnliches.


EDIT: Jetzt, wo ich Ihre Frage besser verstehe, denke ich, dass Sie unnötigerweise zu weit gehen. Wenn Sie einen regulären nehmen SingleThreadExecutor, geben Sie ihm alle Aufgaben, und es wird die Warteschlange nativ erledigen.

Yuval Adam
quelle
Mit SingleThreadExecutor können Sie am besten feststellen, dass alle Threads abgeschlossen sind. Ich habe ein Beispiel gesehen, das eine Weile dauert! Executor.isTerminated, aber das scheint nicht sehr elegant zu sein. Ich habe für jeden Mitarbeiter eine Rückruffunktion implementiert und eine Anzahl erhöht, die funktioniert.
Bär
5

Wenn Sie sicherstellen möchten, dass keine Aufgaben gleichzeitig ausgeführt werden, verwenden Sie einen SingleThreadedExecutor . Die Aufgaben werden in der Reihenfolge bearbeitet, in der sie eingereicht wurden. Sie müssen die Aufgaben nicht einmal halten, sondern sie nur an den Exec senden.

Basszero
quelle
2

Einfacher Code zur Implementierung des CallbackMechanismus mitExecutorService

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

Ausgabe:

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

Wichtige Hinweise:

  1. Wenn Sie Aufgaben nacheinander in FIFO-Reihenfolge bearbeiten möchten, ersetzen Sie sie newFixedThreadPool(5) durchnewFixedThreadPool(1)
  2. Wenn Sie die nächste Aufgabe nach der Analyse des Ergebnisses callbackder vorherigen Aufgabe bearbeiten möchten, entfernen Sie einfach den Kommentar unter der Zeile

    //System.out.println("future status:"+future.get()+":"+future.isDone());
  3. Sie können durch newFixedThreadPool()eines von ersetzen

    Executors.newCachedThreadPool()
    Executors.newWorkStealingPool()
    ThreadPoolExecutor

    abhängig von Ihrem Anwendungsfall.

  4. Wenn Sie die Rückrufmethode asynchron behandeln möchten

    ein. ExecutorService or ThreadPoolExecutorÜbergeben Sie eine freigegebene an Callable-Aufgabe

    b. Konvertieren Sie Ihre CallableMethode in eine Callable/RunnableAufgabe

    c. Rückrufaufgabe auf drücken ExecutorService or ThreadPoolExecutor

Ravindra Babu
quelle
1

Nur um Matts Antwort zu ergänzen, die geholfen hat, hier ein ausführlicheres Beispiel, das die Verwendung eines Rückrufs zeigt.

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

Die Ausgabe ist:

    getPrimeAsync(); prime=241
    onPrimeListener; p=241
    Adios mi amigito
Alter Jack
quelle
1

Sie können eine Implementierung von Callable so verwenden, dass

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

wo CallbackInterface etwas sehr Grundlegendes ist

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

und jetzt wird die Hauptklasse so aussehen

ExecutorService ex = Executors.newFixedThreadPool(2);

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

Dies ist eine Erweiterung von Paches Antwort mit Guavas ListenableFuture.

Insbesondere können Futures.transform()Rückgaben ListenableFutureverwendet werden, um asynchrone Aufrufe zu verketten. Futures.addCallback()gibt zurück void, kann also nicht zum Verketten verwendet werden, ist aber gut für die Behandlung von Erfolg / Misserfolg bei einem asynchronen Abschluss.

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

ANMERKUNG: Zusätzlich zur Verkettung asynchroner Aufgaben Futures.transform()können Sie jede Aufgabe auf einem separaten Executor planen (in diesem Beispiel nicht gezeigt).

bcorso
quelle
Das scheint ziemlich nett zu sein.
Kaiser