使用 compose 创建 CompletableFutures 链表

Question

我想创建一个 CompletableFutures 链。

我正在尝试按以下方式构建内容。

任务 1 执行某些操作，returns 任务 1 完成后结果是一个字符串我想以任务 1 的结果作为输入来启动任务 2。任务 2 准备就绪后 returns 一个整数等等...

所以它应该是非常动态的所以我已经有了这个:

      try {
            CompletableFuture<String> task1Future = CompletableFuture.supplyAsync(new Task1());
            CompletableFuture<String> result = task1Future.thenCompose(task1Result -> CompletableFuture.supplyAsync(new Task2(task1Result)));
            System.out.println(result.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }


        public class Task1 implements Supplier<String> {

    public Task1() {
        System.out.println("Task 1 started");
    }

    @Override
    public String get() {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Result Task 1";
    }
}

我知道尝试实现的是构建一个包装器（一种链表）：

任务列表，其中任务应该是：

public class Task {
    private Supplier startTask;
    private Task followUpTask;

    public Task(Supplier startTask, Task followUpTask) {
        this.startTask = startTask;
        this.followUpTask = followUpTask;
    }
}

但我现在被困住了，因为我不知道如何进行链接以及如何使 Task 更通用，以便可以使用上一个任务的结果启动它。

所以我需要一个方法来构造 CompletableFuture 并只说 start() 一切都会发生。

路上有人可以帮我吗？

Answer 1

您应该将 Task2 定义为函数，因为它接受先前的字符串结果并生成新的整数结果：

public static class Task2 implements Function<String, Integer> {

    public Task2() {
        System.out.println("Task 2 started");
    }

    @Override
    public Integer apply(String s) {
        return s.length();
    }
}

然后您可以按如下方式链接它们：

   CompletableFuture<String> task1Future = CompletableFuture.supplyAsync(new Task1());
   CompletableFuture<Integer> result = task1Future.thenApply(new Task2());
   System.out.println(result.get());

Answer 2

以防万一您希望所有任务都实现 Function，您可以按如下方式启动链：

CompletableFuture<String> task1Future = CompletableFuture.completedFuture("S")
    .thenApply(new Task1());

其中 completedFuture("S") 包含第一个任务的参数：

public static class Task1 implements Function<String, String>  {

    public Task1() {
        System.out.println("Task 1 started");
    }

    @Override
    public String apply(String s) {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Result Task 1";
    }
}

使用 compose 创建 CompletableFutures 链表

Create linked list of CompletableFutures with compose

java

asynchronous

completable-future