Какова лучшая практика, чтобы остановить выполнение цепочки функций при асинхронном выполнении?

Question

У меня есть эта функция здесь:

Function<Integer, Integer> func = (value) -> value + 5;
        func = func.andThen((value) -> {
            //Imagine that here some code executed and raised an exception, I'm throwing it 
            //manually just for the sake of this example.
            throw new RuntimeException("failed");
        });
        func = func.andThen((value) -> {
            System.out.println("Reached last function !");
            return value;
        });
        executeFunction(func);

Теперь вы можете видеть, что я вызываю исключение Runtime в первом методе andThen.Это потому, что я хочу помешать второй и затем выполнить.Это лучший способ сделать это?

Кроме того, я замечаю, что если я выполняю эту функцию внутри другого потока (асинхронно), исключение не выводится в моей консоли, и я хочу знать, что исключение произошло.

private static void executeFunction(Function<Integer, Integer> function) {
        CompletableFuture.supplyAsync(() -> function.apply(100));
    }

В этом случае, если я хочу убедиться, что исключение регистрируется, но что следующая функция в цепочке andThen не выполняется, я должен войти и бросить?Разве это не паттерн ати?

Answer 1

Создание и создание большого количества исключений может стать довольно дорогим, поэтому они должны быть ограничены исключительными обстоятельствами.Вместо этого вы можете использовать Необязательно для потока управления:

func = (value) -> Optional.of(value + 5);
func = func.andThen((optionalValue) -> {
    // Instead of throwing an exception, return an empty Optional
    System.out.println("Log the failure");
    return Optional.empty();
});
func = func.andThen((optionalValue) -> {
    optionalValue.map((value) -> { // This lambda will only execute if optionalValue is not empty
        System.out.println("Reached last function !");
        return value; // map wraps this in an Optional
    });
});
// Finally, unwrap the value. Optional provides a number of ways to do this, depending on how you want to handle failure/empty
func = func.andThen((optional) -> optional.orElse(...));
executeFunction(func);

Answer 2

Вы можете получить желаемое поведение, используя метод CompletableFuture.thenApply.Например:

public class Answer {
  public static void main(String[] args) {
    Function<Integer, Integer> fn0 = v -> v + 5;

    Function<Integer, Integer> fn1 = v -> {
        throw new RuntimeException("failed");
    };

    Function<Integer, Integer> fn2 = v -> {
        System.out.println("Reached last function !");
        return v;
    };

    CompletableFuture.supplyAsync(() -> fn0.apply(100))
            .thenApply(fn1)
            .thenApply(fn2)
            .exceptionally(throwable -> {
                // next line prints the exception thrown by fn1, wrapped in java.util.concurrent.CompletionException
                System.out.println("Failed with error: " + throwable); 
                return 0; // default value, used when exception is thrown
            });
  }
}

В принципе, цепочка CompletableFuture будет прервана исключением "из коробки", поэтому дополнительная обработка не требуется.

В качестве альтернативы, если вы хотите немногоболее общий подход:

public class Answer {
  public static void main(String[] args) {
    executeAsync(() -> stepOne(100))
            .thenApply(Answer::stepTwo)
            .thenApply(Answer::finalStep)
            .exceptionally(Answer::handleException);
  }

  private static CompletableFuture<Integer> executeAsync(Supplier<Integer> task) {
    return CompletableFuture.supplyAsync(task::get);
  }

  private static Integer stepOne(Integer value) {
    return value + 5;
  }

  private static Integer stepTwo(Integer value) {
    throw new RuntimeException("failed");
  }

  private static Integer finalStep(Integer value) {
    System.out.println("Reached last function !");
    return value;
  }

  private static Integer handleException(Throwable throwable) {
    // next line prints the exception thrown by any step before, wrapped in java.util.concurrent.CompletionException
    System.out.println("Failed with error: " + throwable);
    return 0; // default value
  }

Примечания:

• Используя thenApply, вы можете связать столько вызовов функций, сколько вам нужно

• В последнем примере методы внутри того же класса могут быть заменены методами из других классов (не обязательно статических)

Answer 3

Вы можете написать оболочку для функций / runnables, которая регистрирует и завершает работу при сбое вашей задачи.Вот так:

class Runnables
{
    public static Runnable trying(Runnable... runnables)
    {
        return () ->
        {
            int successes = 0;
            try
            {
                for(Runnable runnable : runnables)
                {
                    runnable.run();
                    successes++;
                }
            }
            catch(Throwable t)
            {
                logger.error("Exception thrown from "+successes+"th runnable: ",t);
            }
        };
    }
}

Тогда:

private static void executeFunction(Runnable... runnables)
{
    CompletableFuture.supplyAsync(Runnables.trying(runnables));
}