Wie zu unterbrechen zugrunde liegenden Ausführung von CompletableFuture

Ich weiß, dass CompletableFuture design hat keine Kontrolle über seine Ausführung mit Unterbrechungen, aber ich nehme an, einige von Euch haben dieses problem. CompletableFutures sind sehr gute Art und Weise zu Komponieren asynchrone Ausführung, aber gegeben der Fall, wenn Sie möchten, dass die zugrunde liegenden Ausführung unterbrochen oder gestoppt, wenn die Zukunft wird abgesagt, wie machen wir das? Oder müssen wir einfach akzeptieren, dass alle abgebrochen oder manuell abgeschlossen CompletableFuture wird keine Auswirkungen auf die thread-Entwicklung zu vervollständigen?

Dass ist, meiner Meinung nach, offensichtlich eine nutzlose Arbeit, die Zeit braucht, der Testamentsvollstrecker Arbeiter. Ich Frage mich, welches Konzept oder design könnte in diesem Fall helfen?

UPDATE

Hier ist ein einfacher test für diese

public class SimpleTest {

  @Test
  public void testCompletableFuture() throws Exception {
    CompletableFuture<Void> cf = CompletableFuture.runAsync(()->longOperation());

    bearSleep(1);

    //cf.cancel(true);
    cf.complete(null);

    System.out.println("it should die now already");
    bearSleep(7);
  }

  public static void longOperation(){
    System.out.println("started");
    bearSleep(5);
    System.out.println("completed");
  }

  private static void bearSleep(long seconds){
    try {
      TimeUnit.SECONDS.sleep(seconds);
    } catch (InterruptedException e) {
      System.out.println("OMG!!! Interrupt!!!");
    }
  }
}

Ich Frage mich, ob wir etwas umsetzen, wie seine statische Methode supplyAsync aber mit zusätzlicher Logik, die prüft, ob seine fertig oder abgesagt wird interrupt der thread ausgeführt wurde, für diese Aufgabe...

InformationsquelleAutor vach | 2015-03-12

14

Einen CompletableFuture ist nicht im Zusammenhang mit der asynchronen Aktion, die möglicherweise vervollständigen.

Da (im Gegensatz zu FutureTask) diese Klasse hat keine direkte Kontrolle über die
die Berechnung, die dadurch abgeschlossen werden, ist eine Stornierung behandelt
nur eine andere form von außergewöhnlichen Abschluss. Methode cancel hat die
gleiche Wirkung wie completeExceptionally(new CancellationException()).

Kann es auch nicht werden einem separaten thread arbeiten abgeschlossen haben (es kann sogar sein, dass viele threads arbeiten daran). Auch wenn es heißt, es gibt keinen link von einer CompletableFuture zu jedem thread, der hat eine Referenz darauf.

Als solche, es gibt nichts, was Sie tun können, durch CompletableFuture zu unterbrechen, alle Threads, die ausgeführt werden können, manche Aufgabe, die werden es vollenden. Sie müssen schreiben Sie Ihre eigene Logik, welche Spuren jede Thread Instanzen, die erfassen einer Referenz auf die CompletableFuture mit der Absicht, es zu vollenden.

Hier ist ein Beispiel für die Art der Ausführung, die ich denke, Sie bekommen konnte Weg mit.
```
public static void main(String[] args) throws Exception {
    ExecutorService service = Executors.newFixedThreadPool(1);
    CompletableFuture<String> completable = new CompletableFuture<>();
    Future<?> future = service.submit(new Runnable() {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                if (Thread.interrupted()) {
                    return; //remains uncompleted
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    return; //remains uncompleted
                }
            }
            completable.complete("done");
        }
    });

    Thread.sleep(2000);

    //not atomic across the two
    boolean cancelled = future.cancel(true);
    if (cancelled)
        completable.cancel(true); //may not have been cancelled if execution has already completed
    if (completable.isCancelled()) {
        System.out.println("cancelled");
    } else if (completable.isCompletedExceptionally()) {
        System.out.println("exception");
    } else {
        System.out.println("success");
    }
    service.shutdown();
}
```
Dies setzt Voraus, dass die Aufgabe ausgeführt wird, ist setup zu handhaben Unterbrechungen korrekt.
- gut ich wusste das, aber meine Frage ist über diese Logik zu verknüpfen... zum Beispiel nehmen wir an, ich werde meine eigene statische factory-Methode verwendet für diese Fälle... statt supplyAsync ich werde einige benutzerdefinierte factory-Methode... dont Sie denken, es ist möglich, link CF-thread, der führt es...?
- Sie müssten, um zu garantieren, dass die nur Threads, erhalten Sie Zugang zu den CompletableFuture sind diejenigen, die durch Ihre benutzerdefinierte factory-Methode. Das ist nicht sehr praktikabel, imo. Die wichtige Sache hier ist, dass die CF ist nicht im Zusammenhang mit einer Thread. Wenn Sie stornieren müssen einige thread ' s funktioniert, sollten Sie vorgelegt haben, die die Arbeit durch eine ExecutorService abgerufen werden die Future und abgebrochen dass. Die Runnable oder Callable eingereicht haben, können Sie einen Verweis auf die VGL. Sie werden nicht zur Aufhebung der CF, werden Sie das Abbrechen der Aufgabe durch die Future.
- Sie können Kapseln die Logik durch den Beitritt der Future und CompletableFuture und Kündigung von beiden. Beachten Sie, dass dies nicht atomar.
- Sache ist, dass ich brauchen Flexibilität in der Verkettung der Vorgänge, die ausfüllbaren futures bieten, aber ich dont wollen, dass Sie lassen Sie nutzlose Aktivitäten weiter zu Taille mein cpu-Zyklen, wenn ich Sie absagen... das ist sinnlos..
- also ich kann CompletableFuture die Zukunft aus? und bei CF ist abgesagt, rufen Sie die gleichen absagen der Zukunft? Könnten Sie bitte einen Anhaltspunkt geben, wie das zu tun?
- Ich bin sogar bereit, erstreckt CF und fügen Abhängigkeit zu F, wenn das ist nur Weg 🙁
- Ich dachte, dass RxJava wird, um dieses problem in der Regel scheint aber das gleiche problem gibt es auch...
- Ich kann mir vorstellen denken Sie an einen Fall, wo Sie runAsync und Abbrechen möchten, die Aufgabe? Das ist jedoch nicht der einzige use-case für CompletableFuture. Sie starten die 10 threads die alle Arbeit in Richtung Abschluss. Unterbrechen diesen threads ist bis zu Ihnen. Die Fäden selbst können nicht wissen, Sie haben einen Verweis auf eine CF und die CF definitiv nicht wissen können. Nur der Entwickler kann.
- Erweiterung Future und CompletableFuture wird nicht sehr sein nützlich, wenn Sie planen, JDK-Bibliotheken zur Abgabe der Arbeit (ExecutorServices und so).
- Ich nicht wirklich sehen, warum mit ExecutorService nicht machen es nützlich... das problem ist, dass code ausgeführt werden könnten, die die gleiche Aufgabe 100-mal in einer Sekunde und ich will diese 99 zu stornieren, wenn Sie nicht machen es in der Zeit... ja, ich denke, das ist nicht zumutbar zu lassen, einige sinnlos-thread verbrauchen Ressourcen...
- Ich Stimme völlig mit Ihnen. Vergeuden Sie nicht Ressourcen. Aber Sie werden nicht in der Lage, dies zu lösen, mit den bestehenden CF. Sie können erweitern, CF und wickeln Sie ein Future, aber ich glaube nicht, dass Sie ' ll haben Zugriff auf eine Menge der Komponenten, die mit der Fertigstellung (eine Menge ist es private) und ich bezweifle, dass Sie werde in der Lage sein durchführen die Stornierung atomar.
- Ich werde schreiben, ein Beispiel, was ich meine. Geben Sie mir ein paar.
- das ist es, was ich fürchtete, müssen digg diesen code und das Rad erfinden... ich Frage mich, wie Leute nicht dieses problem hatte... Ok warten 🙂
- Aktualisiert mit einem extrem einfachen Beispiel mit nur einem einzigen thread der Ausführung von arbeiten, die möglicherweise den CompletableFuture.
- Vielen Dank für das Beispiel, aber Sie sind cancling die Zukunft nicht die ausfüllbaren... mein Ziel ist es, die completablefuture.Abbrechen unterbrechen zugrunde liegenden thread, wenn Ihr nicht fertig... oder habe ich da etwas falsch verstanden?
- Eventutally werde ich über eine factory-Methode selbst gibt CompletableFuture (aber zugrunde liegenden Implementierung könnte meine eigenen), das unterbricht den thread und wenn es einer ist, wenn es abgesagt... das bedeutet ich muss irgendwie CF bewusst sein, dass thread, oder Rückruf, der wird feststellen, dass die thread-und interrupt-call-beim CF ist abgebrochen...
- Damit Sie Ihre genaue Ziel ist unerreichbar, da es keine zugrunde liegenden threads zu einem CompletableFuture. Der Verein, den du machst zwischen deinen threads und einer CompletableFuture ganz alleine. CompletableFuture war nicht so vorgesehen. Die nächsten wirst du mit einem Future und ein CompletableFuture ist, was ich oben dargestellt. Abbrechen der Future. Wenn es erfolgreich war, kündigen die damit verbundenen CompletableFuture. Offensichtlich, Sie könnten in der Lage sein, um das gewünschte Verhalten durch das schreiben Ihrer eigenen CompletableFuture, aber das ist keine leichte Aufgabe.
- Vielen Dank für deine Erklärung. In Ihrem Beispiel verwenden Sie Thread.sleep() welche bereits überprüft, die für die Unterbrechung, also mit Thread.sleep() vielleicht nicht die bessere Möglichkeit zu erklären, Ihren Standpunkt (auch wenn die Erklärung richtig ist). Zum Beispiel, Kommentar if (Thread.interrupted()) {..} und das Programm wird unterbrochen, genau das gleiche. Ich denke, statt Thread.sleep() den code Lesen sollte doIninterrumpibleComputation().
InformationsquelleAutor Sotirios Delimanolis

Was ist das?

public static <T> CompletableFuture<T> supplyAsync(final Supplier<T> supplier) {

    final ExecutorService executorService = Executors.newFixedThreadPool(1);

    final CompletableFuture<T> cf = new CompletableFuture<T>() {
        @Override
        public boolean complete(T value) {
            if (isDone()) {
                return false;
            }
            executorService.shutdownNow();
            return super.complete(value);
        }

        @Override
        public boolean completeExceptionally(Throwable ex) {
            if (isDone()) {
                return false;
            }
            executorService.shutdownNow();
            return super.completeExceptionally(ex);
        }
    };

    //submit task
    executorService.submit(() -> {
        try {
            cf.complete(supplier.get());
        } catch (Throwable ex) {
            cf.completeExceptionally(ex);
        }
    });

    return cf;
}

Einfachen Test:

    CompletableFuture<String> cf = supplyAsync(() -> {
        try {
            Thread.sleep(1000L);
        } catch (Exception e) {
            System.out.println("got interrupted");
            return "got interrupted";
        }
        System.out.println("normal complete");
        return "normal complete";
    });

    cf.complete("manual complete");
    System.out.println(cf.get());

Ich weiß nicht, wie die Idee, dass Sie einen Testamentsvollstrecker service, jedes mal, aber vielleicht können Sie einen Weg finden, um die Wiederverwendung der ForkJoinPool.

Froh, dass Sie es nützlich finden. Wenn das, was Sie versuchen zu erreichen, obwohl kann umgesetzt werden über einen timeout-Mechanismus, vielleicht findest du diese Lösung noch interessanter: stackoverflow.com/questions/23575067/...
Versteh mich nicht falsch, ich interessiere mich nicht wirklich, aber könnten Sie mir sagen, warum haben Sie verändert die richtige Antwort diese Frage?
Sicher, dass Ruben, nach der Verwendung dieser für eine Weile habe ich verstanden, dass die CF ' s sind nicht dafür ausgelegt und wenn sich jemand über zu tun, was ich suchte, dann sind Sie missbrauchen, die Klasse und muss wohl lösen das problem auf eine andere Weise. Aber vielen Dank für die Umsetzung der eigentlichen Lösung, die ich Schätze, und ich lernte eine Menge von Proben ähnlich. Ändern Sie einfach die richtige Antwort, weil ich jetzt denke, es ist besser, dies nicht zu tun, und verwenden Sie richtigen tools für die richtigen Dinge 🙂

InformationsquelleAutor Ruben

1

Bitte Lesen Sie meine Antwort auf die Frage:
Transformation von Java-Zukunft in einem CompletableFuture

In den code dort bereits erwähnt, die CompletionStage Verhalten Hinzugefügt wird RunnableFuture Unterklasse (von ExecutorService-Implementierungen), so können Sie unterbrechen, es in der richtigen Weise.
- BTW, ich veröffentlicht meinen code als separate Bibliothek - bitte siehe github.com/vsilaev/tascalate-concurrent
InformationsquelleAutor Valery Silaev

Was?

/** @return {@link CompletableFuture} which when cancelled will interrupt the supplier
 */
public static <T> CompletableFuture<T> supplyAsyncInterruptibly(Supplier<T> supplier, Executor executor) {
    return produceInterruptibleCompletableFuture((s) -> CompletableFuture.supplyAsync(s, executor), supplier);
}

//in case we want to do the same for similar methods later
private static <T> CompletableFuture<T> produceInterruptibleCompletableFuture(
        Function<Supplier<T>,CompletableFuture<T>> completableFutureAsyncSupplier, Supplier<T> action) {
    FutureTask<T> task = new FutureTask<>(action::get);
    return addCancellationAction(completableFutureAsyncSupplier.apply(asSupplier(task)), () ->
            task.cancel(true));
}

/** Ensures the specified action is executed if the given {@link CompletableFuture} is cancelled.
 */
public static <T> CompletableFuture<T> addCancellationAction(CompletableFuture<T> completableFuture,
                                                             @NonNull Runnable onCancellationAction) {
    completableFuture.whenComplete((result, throwable) -> {
        if (completableFuture.isCancelled()) {
            onCancellationAction.run();
        }
    });
    return completableFuture;  //return original CompletableFuture
}

/** @return {@link Supplier} wrapper for the given {@link RunnableFuture} which calls {@link RunnableFuture#run()}
 *          followed by {@link RunnableFuture#get()}.
 */
public static <T> Supplier<T> asSupplier(RunnableFuture<T> futureTask) throws CompletionException {
    return () -> {
        try {
            futureTask.run();
            try {
                return futureTask.get();
            } catch (ExecutionException e) {  //unwrap ExecutionExceptions
                final Throwable cause = e.getCause();
                throw (cause != null) ? cause : e;
            }
        } catch (CompletionException e) {
            throw e;
        } catch (Throwable t) {
            throw new CompletionException(t);
        }
    };
}

InformationsquelleAutor user6519354

0

Wenn Sie
```
cf.get();
```
statt
```
cf.join();
```
Der thread wartet auf die Beendigung unterbrochen werden kann. Dieser Biss mich in den a**, also ich bin gerade dabei es gibt. Sie würden dann weitergeben müssen, diese Unterbrechung weiter verwenden /vgl.Abbrechen (...), um wirklich zu beenden der Ausführung.

InformationsquelleAutor jontejj

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.