Превратите будущее Java в CompletableFuture

97

В Java 8 представлена CompletableFutureновая реализация Future, которую можно компоновать (включает набор методов thenXxx). Я хотел бы использовать только это, но многие библиотеки, которые я хочу использовать, возвращают только несоставные Futureэкземпляры.

Есть ли способ обернуть возвращенные Futureэкземпляры внутри, CompleteableFutureчтобы я мог их скомпоновать?

Дэн Мидвуд
источник

Ответы:

57

Способ есть, но он тебе не понравится. Следующий метод преобразует a Future<T>в a CompletableFuture<T>:

public static <T> CompletableFuture<T> makeCompletableFuture(Future<T> future) {
  if (future.isDone())
    return transformDoneFuture(future);
  return CompletableFuture.supplyAsync(() -> {
    try {
      if (!future.isDone())
        awaitFutureIsDoneInForkJoinPool(future);
      return future.get();
    } catch (ExecutionException e) {
      throw new RuntimeException(e);
    } catch (InterruptedException e) {
      // Normally, this should never happen inside ForkJoinPool
      Thread.currentThread().interrupt();
      // Add the following statement if the future doesn't have side effects
      // future.cancel(true);
      throw new RuntimeException(e);
    }
  });
}

private static <T> CompletableFuture<T> transformDoneFuture(Future<T> future) {
  CompletableFuture<T> cf = new CompletableFuture<>();
  T result;
  try {
    result = future.get();
  } catch (Throwable ex) {
    cf.completeExceptionally(ex);
    return cf;
  }
  cf.complete(result);
  return cf;
}

private static void awaitFutureIsDoneInForkJoinPool(Future<?> future)
    throws InterruptedException {
  ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    @Override public boolean block() throws InterruptedException {
      try {
        future.get();
      } catch (ExecutionException e) {
        throw new RuntimeException(e);
      }
      return true;
    }
    @Override public boolean isReleasable() {
      return future.isDone();
    }
  });
}

Очевидно, проблема с этим подходом состоит в том, что для каждого Future поток будет заблокирован, чтобы ждать результата Future, что противоречит идее Futures. В некоторых случаях можно было бы добиться большего. Однако в целом без активного ожидания результата в Будущем нет решения .

носид
источник
1
Ха, это именно то, что я написал, прежде чем подумать, что должен быть лучший способ. Но, думаю, нет
Дэн Мидвуд
12
Хммм ... разве это решение не съедает одну из нитей "общего пула", просто для ожидания? Эти потоки "общего пула" никогда не должны блокироваться ... хммм ...
Пети
1
@Peti: Ты прав. Однако суть в том, что если вы, скорее всего, делаете что-то неправильно, независимо от того, используете ли вы общий пул или неограниченный пул потоков .
nosid 04
4
Это может быть не идеально, но использование CompletableFuture.supplyAsync(supplier, new SinglethreadExecutor())по крайней мере не блокирует потоки общего пула.
MikeFHay
6
Пожалуйста, просто никогда не делайте этого
Laymain
56

Если библиотека, которую вы хотите использовать, также предлагает метод стиля обратного вызова в дополнение к стилю Future, вы можете предоставить ей обработчик, который завершает CompletableFuture без какой-либо дополнительной блокировки потоков. Вот так:

    AsynchronousFileChannel open = AsynchronousFileChannel.open(Paths.get("/some/file"));
    // ... 
    CompletableFuture<ByteBuffer> completableFuture = new CompletableFuture<ByteBuffer>();
    open.read(buffer, position, null, new CompletionHandler<Integer, Void>() {
        @Override
        public void completed(Integer result, Void attachment) {
            completableFuture.complete(buffer);
        }

        @Override
        public void failed(Throwable exc, Void attachment) {
            completableFuture.completeExceptionally(exc);
        }
    });
    completableFuture.thenApply(...)

Я вижу, что без обратного вызова единственный другой способ решить эту проблему - использовать цикл опроса, который помещает все ваши Future.isDone()проверки в один поток, а затем вызывает завершение всякий раз, когда можно получить Future.

Кафкианский
источник
1
Я использую асинхронную библиотеку Apache Http, которая принимает FutureCallback. Это облегчило мне жизнь :)
Абхишек Гаяквад
14

Если ваш Future- результат вызова ExecutorServiceметода (например submit()), проще всего использовать этот CompletableFuture.runAsync(Runnable, Executor)метод.

Из

Runnbale myTask = ... ;
Future<?> future = myExecutor.submit(myTask);

к

Runnbale myTask = ... ;
CompletableFuture<?> future = CompletableFuture.runAsync(myTask, myExecutor);

CompletableFutureЗатем создается «родной».

РЕДАКТИРОВАТЬ: следуя комментариям @SamMefford, исправленным @MartinAndersson, если вы хотите передать a Callable, вам нужно позвонить supplyAsync(), преобразовав Callable<T>в a Supplier<T>, например, с помощью:

CompletableFuture.supplyAsync(() -> {
    try { return myCallable.call(); }
    catch (Exception ex) { throw new RuntimeException(ex); } // Or return default value
}, myExecutor);

Поскольку T Callable.call() throws Exception;выдает исключение и T Supplier.get();не делает этого, вы должны перехватить исключение, чтобы прототипы были совместимы.

Матье
источник
1
Или, если вы используете Callable <T>, а не Runnable, вместо этого попробуйте supplyAsync:CompletableFuture<T> future = CompletableFuture.supplyAsync(myCallable, myExecutor);
Сэм Меффорд,
@SamMefford, спасибо, я отредактировал, чтобы включить эту информацию.
Матье
supplyAsyncполучает Supplier. Код не будет компилироваться, если вы попытаетесь передать файл Callable.
Мартин Андерссон,
@MartinAndersson, верно, спасибо. Я редактировал дальше, чтобы преобразовать a Callable<T>в Supplier<T>.
Матье
10

Я опубликовал небольшой проект будущего, который пытается сделать лучше, чем простой способ ответа.

Основная идея состоит в том, чтобы использовать только один поток (и, конечно же, не только с циклом вращения) для проверки всех состояний Futures внутри, что помогает избежать блокировки потока из пула для каждого преобразования Future -> CompletableFuture.

Пример использования:

Future oldFuture = ...;
CompletableFuture profit = Futurity.shift(oldFuture);
Дмитрий Спихальский
источник
Это выглядит интересно. Он использует поток таймера? Почему это не принятый ответ?
Кира
@Kira Да, он в основном использует один поток таймера для ожидания всех отправленных фьючерсов.
Дмитрий Спихальский,
8

Предложение:

http://www.thedevpiece.com/converting-old-java-future-to-completablefuture/

Но, в основном:

public class CompletablePromiseContext {
    private static final ScheduledExecutorService SERVICE = Executors.newSingleThreadScheduledExecutor();

    public static void schedule(Runnable r) {
        SERVICE.schedule(r, 1, TimeUnit.MILLISECONDS);
    }
}

И CompletablePromise:

public class CompletablePromise<V> extends CompletableFuture<V> {
    private Future<V> future;

    public CompletablePromise(Future<V> future) {
        this.future = future;
        CompletablePromiseContext.schedule(this::tryToComplete);
    }

    private void tryToComplete() {
        if (future.isDone()) {
            try {
                complete(future.get());
            } catch (InterruptedException e) {
                completeExceptionally(e);
            } catch (ExecutionException e) {
                completeExceptionally(e.getCause());
            }
            return;
        }

        if (future.isCancelled()) {
            cancel(true);
            return;
        }

        CompletablePromiseContext.schedule(this::tryToComplete);
    }
}

Пример:

public class Main {
    public static void main(String[] args) {
        final ExecutorService service = Executors.newSingleThreadExecutor();
        final Future<String> stringFuture = service.submit(() -> "success");
        final CompletableFuture<String> completableFuture = new CompletablePromise<>(stringFuture);

        completableFuture.whenComplete((result, failure) -> {
            System.out.println(result);
        });
    }
}
Габриэль Франциско
источник
это довольно просто рассуждать о & элегантном и подходящем для большинства случаев использования. Я бы сделал CompletablePromiseContext нестатический и взял параметр для интервала проверки (который здесь установлен на 1 мс), а затем перегрузил CompletablePromise<V>конструктор, чтобы иметь возможность предоставить свой собственный, CompletablePromiseContextвозможно, с другим (более длинным) интервалом проверки для длительного времени, Future<V>когда вы не 'не обязательно иметь возможность запускать обратный вызов (или создавать) сразу после завершения, и вы также можете иметь экземпляр CompletablePromiseContextдля просмотра набора Future(если у вас их много)
Декстер Легаспи,
5

Позвольте мне предложить другой (надеюсь, лучший) вариант: https://github.com/vsilaev/java-async-await/tree/master/com.farata.lang.async.examples/src/main/java/com/farata / одновременный

Вкратце идея такая:

  1. Представляем CompletableTask<V>интерфейс - объединение CompletionStage<V>+RunnableFuture<V>
  2. Деформация ExecutorServiceдля возврата CompletableTaskиз submit(...)методов (вместо Future<V>)
  3. Готово, у нас есть запускаемые И составные фьючерсы.

Реализация использует альтернативную реализацию CompletionStage (обратите внимание, CompletionStage, а не CompletableFuture):

Применение:

J8ExecutorService exec = J8Executors.newCachedThreadPool();
CompletionStage<String> = exec
   .submit( someCallableA )
   .thenCombineAsync( exec.submit(someCallableB), (a, b) -> a + " " + b)
   .thenCombine( exec.submit(someCallableC), (ab, b) -> ab + " " + c); 
Валерий Силаев
источник
2
Небольшое обновление: код перенесен в отдельный проект, github.com/vsilaev/tascalate-concurrent , и теперь можно использовать нестандартные Executor-ы из java.util.concurrent.
Валерий Силаев