Java 8 CompletableFuture

оригинал:Java 8 CompletableFutures Part I

Автор: Билл Беджек
Переводчик: никто

Предисловие переводчика

Добавлен JDK1.5FutureИнтерфейс, но использование интерфейса не очень подходит для нужд асинхронной разработки, да и пользоваться им не так удобно. Так что есть много сторонних пакетовFuture, Гуава обеспечивает лучшееListenableFutureкласс, а Netty предоставляет один из своихFuture. Итак, в Java8CompletableFutureтак сказатьFutureС некоторыми болями вы можете элегантно выполнять составное асинхронное программирование, и в то же время оно больше соответствует функциональному программированию.

Java 8 отсутствовала долгое время, и в нее был добавлен отличный инструмент управления параллелизмом, которыйCompletableFutureДобрый.CompletableFutureДостигнутоFutureинтерфейса, и он может задавать значения явно, что более интересно, мы можем обрабатывать цепочки и поддерживать зависимые поведения, которые определяютсяCompletableFutureАктивируется завершением.CompletableFutureПохоже на гуавуListenableFutureДобрый. Оба они обеспечивают схожую функциональность и не будут сравниваться в этой статье. я был в предыдущемстатьявведен вListenableFutrue. Хотя дляListenableFutrueВведение немного устарело, но подавляющее большинство знаний все еще применимо.CompletableFutureДокументация уже очень обширна, но в ней отсутствуют конкретные примеры того, как их использовать. Эта статья предназначена для того, чтобы показать, как использовать серию простых примеров в модульном тестировании.CompletableFuture. Изначально хотел закончить в одной статьеCompleteableFuture, но информации так много, что кажется лучше разделить на три части:

Создавайте/составляйте задачи и добавляйте к ним слушателей.
Обработка ошибок и восстановление после ошибок.
Отменить или принудительно завершить.

Начало работы с CompletableFuture

НачинаяCompletableFutureПеред этим нам потребуются некоторые базовые знания.CompletableFutureДостигнутоCompletionStageинтерфейс. Кратко описано в javadocCompletionStage:

Возможно, асинхронно вычисляемый этап, выполняющий операцию или вычисляющий значение после завершения другого CompletionStage. Завершение этапа зависит от результата его собственного расчета, который, в свою очередь, может инициировать другие зависимые этапы.

Полная документация CompletionStage содержит большое количество контента, поэтому мы суммируем здесь несколько ключевых моментов:

можно рассчитать поFuture,Consumer или Runnableв интерфейсеapply,accept или runвыражается другими способами.
Выполнение расчета в основном включает в себя следующее

А. Выполнение по умолчанию (может вызвать поток)

б. Используйте значение по умолчаниюCompletionStageПоставщик асинхронного выполнения для асинхронного выполнения. Эти имена методов используютsomeActionAsyncЭтот формат представляет.

в. использоватьExecutorПоставщик выполняется асинхронно. Эти методы такжеsomeActionAsyncэтот формат, но добавляетExecutorпараметр.

Далее я буду цитировать непосредственно в этой статьеCompletableFuture а также CompletionStage.

Создать CompleteableFuture

СоздаватьCompleteableFutureПросто, но не очень понятно. Самый простой способ - использоватьCompleteableFuture.completedFutureметод, который возвращает новый и завершенныйCompleteableFuture:

@Test
public void test_completed_future() throws Exception {
  String expectedValue = "the expected value";
  CompletableFuture<String> alreadyCompleted = CompletableFuture.completedFuture(expectedValue);
  assertThat(alreadyCompleted.get(), is(expectedValue));
}

Это выглядит немного скучно, позже мы увидим, как создать завершенныйCompleteableFutureпригодится.

Теперь давайте посмотрим, как создать представление асинхронной задачи.CompleteableFuture:

private static ExecutorService service = Executors.newCachedThreadPool();

@Test
public void test_run_async() throws Exception {
    CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> 		                   System.out.println("running async task"), service);
    //utility testing method
    pauseSeconds(1);
    assertThat(runAsync.isDone(), is(true));
}

@Test
public void test_supply_async() throws Exception {
    CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(simulatedTask(1, "Final Result"), service);
    assertThat(completableFuture.get(), is("Final Result"));
}

В первом способе мы видимrunAsyncзадача, во втором способе, тоsupplyAsyncпример. Это может быть очевидно, однако использованиеrunAsyncили использоватьsupplyAsync, в зависимости от того, имеет ли задача возвращаемое значение. В обоих примерах мы предоставляем пользовательскийExecutor, который действует как поставщик асинхронного выполнения. когда используешьsupplyAsyncметод, я лично думаю, используяCallableвместоSupplierЭто кажется более естественным. потому что онифункциональный интерфейс,Callableболее тесно связан с асинхронными задачами и также может генерировать проверенные исключения, в то время какSupplierто нет (хотя мы можем передать небольшое количество кодаПозволятьSupplierгенерировать проверенное исключение).

добавить слушателя

Теперь мы можем создатьCompleteableFutureОбъекты. Чтобы запустить асинхронные задачи, давайте начнем учиться завершать задачи «аудирования» и выполнять некоторые из следующих движений. Здесь вы следуете, когда вы увеличиваетеCompletionStageКогда объект отслеживается, предыдущая задача должна быть полностью успешной, прежде чем можно будет запустить последующие задачи и этапы. В этой статье будут представлены некоторые методы работы с неудавшимися задачами, в то время как вCompleteableFutureРешение для работы с ошибками в цепочке будет представлено в следующих статьях.

 @Test
public void test_then_run_async() throws Exception {
	Map<String,String> cache = new HashMap<>();
    cache.put("key","value");
    CompletableFuture<String> taskUsingCache =           CompletableFuture.supplyAsync(simulatedTask(1,cache.get("key")),service);
    CompletableFuture<Void> cleanUp = taskUsingCache.thenRunAsync(cache::clear,service);
    cleanUp.get();
    String theValue = taskUsingCache.get();
    assertThat(cache.isEmpty(),is(true));
    assertThat(theValue,is("value"));
}

Этот пример в основном показан в первомCompletableFutureПосле успешного завершения запустите задачу очистки. В предыдущем примере, когда начальная задача завершилась успешно, мы использовалиRunnableвыполнение задачи. Мы также можем определить последующую задачу, которая может напрямую получить успешный результат предыдущей задачи.

@Test
public void test_accept_result() throws Exception {
    CompletableFuture<String> task = CompletableFuture.supplyAsync(simulatedTask(1, "add when done"), service);
    CompletableFuture<Void> acceptingTask = task.thenAccept(results::add);
    pauseSeconds(2);
    assertThat(acceptingTask.isDone(), is(true));
    assertThat(results.size(), is(1));
    assertThat(results.contains("add when done"), is(true));
}

это использованиеAcceptПример метода, который получаетCompletableFuture, затем передать результатConsumerобъект. В Java 8,ConsumerУ экземпляров нет возвращаемого значения. Если вы хотите получить побочные эффекты запуска, вам нужно поместить результаты в список.

композиция и композиционные задачи

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

Композиционное задание

Составление означает получение успешногоCompletableFutureрезультат в качестве входных данных черезFunctionвернуть другойCompletableFuture. Ниже приведено использованиеCompletableFuture.thenComposeAsyncпример:

@Test
public void test_then_compose() throws Exception {
    Function<Integer,Supplier<List<Integer>>> getFirstTenMultiples = num ->
                ()->Stream.iterate(num, i -> i + num).limit(10).collect(Collectors.toList());

    Supplier<List<Integer>> multiplesSupplier = getFirstTenMultiples.apply(13);

    //Original CompletionStage

    CompletableFuture<List<Integer>> getMultiples = CompletableFuture.supplyAsync(multiplesSupplier, service);

    //Function that takes input from orignal CompletionStage
    Function<List<Integer>, CompletableFuture<Integer>> sumNumbers = multiples ->
            CompletableFuture.supplyAsync(() -> multiples.stream().mapToInt(Integer::intValue).sum());

    //The final CompletableFuture composed of previous two.

    CompletableFuture<Integer> summedMultiples = getMultiples.thenComposeAsync(sumNumbers, service);
    assertThat(summedMultiples.get(), is(715));
}

В этом примере первыйCompletionStageПриводится список, содержащий 10 чисел, каждое из которых умножается на 13. это обеспечиваетFunctionполучить эти результаты и создать другойCompletionStage, который суммирует числа в списке.

Объединяйте задачи

Комбинированная задача завершается получением двух успешныхCompletionStagesИ вывестиBiFunctionПараметры типа, которые, в свою очередь, дают дополнительные результаты. Ниже приведен очень простой пример, иллюстрирующий комбинациюCompletionStagesчтобы получить результаты.

@Test
public void test_then_combine_async() throws Exception {
    CompletableFuture<String> firstTask = CompletableFuture.supplyAsync(simulatedTask(3, "combine all"), service);

    CompletableFuture<String> secondTask = CompletableFuture.supplyAsync(simulatedTask(2, "task results"), service);

    CompletableFuture<String> combined = firstTask.thenCombineAsync(secondTask, (f, s) -> f + " " + s, service);

    assertThat(combined.get(), is("combine all task results"));
}

В этом примере показано, как объединить две асинхронные задачи.CompletionStage, однако мы также можем объединить уже сделанноеCompletableFutureасинхронная задача. Объединение известного значения, которое необходимо вычислить, также является хорошим способом сделать это:

@Test
public void test_then_combine_with_one_supplied_value() throws Exception {
    CompletableFuture<String> asyncComputedValue = CompletableFuture.supplyAsync(simulatedTask(2, "calculated value"), service);
    CompletableFuture<String> knowValueToCombine = CompletableFuture.completedFuture("known value");

    BinaryOperator<String> calcResults = (f, s) -> "taking a " + f + " then adding a " + s;
    CompletableFuture<String> combined = asyncComputedValue.thenCombine(knowValueToCombine, calcResults);

    assertThat(combined.get(), is("taking a calculated value then adding a known value"));
}

Наконец, это использованиеCompletableFuture.runAfterbothAsyncпример

@Test
public void test_run_after_both() throws Exception {
    CompletableFuture<Void> run1 = CompletableFuture.runAsync(() -> {
        pauseSeconds(2);
        results.add("first task");
    }, service);

    CompletableFuture<Void> run2 = CompletableFuture.runAsync(() -> {
        pauseSeconds(3);
        results.add("second task");
    }, service);

    CompletableFuture<Void> finisher = run1.runAfterBothAsync(run2,() -> results. add(results.get(0)+ "&"+results.get(1)),service);
    pauseSeconds(4);
    assertThat(finisher.isDone(),is(true));
    assertThat(results.get(2),is("first task&second task"));
}

Слушайте первое законченное задание

Во всех предыдущих примерах все результаты должны ждать всехCompletionStageКонец, однако, в том, что спрос не всегда такой. Возможно, нам понадобится получить результат первой выполненной задачи. В следующем примере показано использованиеConsumerПринять первый завершенный результат:

@Test
public void test_accept_either_async_nested_finishes_first() throws Exception {
    CompletableFuture<String> callingCompletable = CompletableFuture.supplyAsync(simulatedTask(2, "calling"), service);
    CompletableFuture<String> nestedCompletable = CompletableFuture.supplyAsync(simulatedTask(1, "nested"), service);

    CompletableFuture<Void> collector = callingCompletable.acceptEither(nestedCompletable, results::add);

    pauseSeconds(2);
    assertThat(collector.isDone(), is(true));
    assertThat(results.size(), is(1));
    assertThat(results.contains("nested"), is(true));
}

аналогичная функцияCompletableFuture.runAfterEither

@Test
public void test_run_after_either() throws Exception {
    CompletableFuture<Void> run1 = CompletableFuture.runAsync(() -> {
            pauseSeconds(2);
            results.add("should be first");
    }, service);

    CompletableFuture<Void> run2 = CompletableFuture.runAsync(() -> {
            pauseSeconds(3);
            results.add("should be second");
    }, service);

    CompletableFuture<Void> finisher = run1.runAfterEitherAsync(run2,() -> results.add(results.get(0).toUpperCase()),service);

    pauseSeconds(4);
    assertThat(finisher.isDone(),is(true));
    assertThat(results.get(1),is("SHOULD BE FIRST"));
 }

несколько комбинаций

Пока все комбинации/конфигурации имеют только два примера.CompletableFutureобъект. Это сделано намеренно, чтобы сделать пример максимально простым и понятным. Мы можем объединить любое количествоCompletionStage. Обратите внимание, что приведенные ниже примеры приведены только для иллюстрации!

@Test
public void test_several_stage_combinations() throws Exception {
    Function<String,CompletableFuture<String>> upperCaseFunction = s -> CompletableFuture.completedFuture(s.toUpperCase());

    CompletableFuture<String> stage1 = CompletableFuture.completedFuture("the quick ");

    CompletableFuture<String> stage2 = CompletableFuture.completedFuture("brown fox ");

    CompletableFuture<String> stage3 = stage1.thenCombine(stage2,(s1,s2) -> s1+s2);

    CompletableFuture<String> stage4 = stage3.thenCompose(upperCaseFunction);

    CompletableFuture<String> stage5 = CompletableFuture.supplyAsync(simulatedTask(2,"jumped over"));

    CompletableFuture<String> stage6 = stage4.thenCombineAsync(stage5,(s1,s2)-> s1+s2,service);

    CompletableFuture<String> stage6_sub_1_slow = CompletableFuture.supplyAsync(simulatedTask(4,"fell into"));

     CompletableFuture<String> stage7 = stage6.applyToEitherAsync(stage6_sub_1_slow,String::toUpperCase,service);

     CompletableFuture<String> stage8 = CompletableFuture.supplyAsync(simulatedTask(3," the lazy dog"),service);

     CompletableFuture<String> finalStage = stage7.thenCombineAsync(stage8,(s1,s2)-> s1+s2,service);

     assertThat(finalStage.get(),is("THE QUICK BROWN FOX JUMPED OVER the lazy dog"));
}

Обратите внимание, что порядок не гарантируется при объединении CompletionStages. В этих модульных тестах предоставляется время для моделирования задач, чтобы обеспечить порядок выполнения.

резюме

В этой статье в основном используетсяCompletableFutureПервая часть класса. В последующих статьях мы в основном будем вводить обработку ошибок и восстановление, принудительное завершение или отмену.

Подписывайтесь на меня: