java - 有没有办法在 Spring Boot 微服务中完成异步进程响应时处理它们？

Question

假设我们在一个函数中有 5 个不同的服务调用，并且都是异步的，我想在它们完成时处理它们。

例如，这些是 3 个不同类中可用的 3 个不同函数

@Async
public CompletableFuture<Actor> lookForId(String id) throws InterruptedException {
    LOG.info("Looking up Movie ID: {}", Id);
    String url = "http://localhost:7073/data/" + id;
    Actor results = restTemplate.getForObject(url, Actors.class);
    // Artificial delay of 1s for demonstration purposes
    Thread.sleep(1000L);
    return CompletableFuture.completedFuture(results);
}

@Async
public CompletableFuture<Singer> lookForId(String id) throws InterruptedException {
    LOG.info("Looking up Movie ID: {}", id);
    String url = "http://localhost:7075/data/" + id;
    Singer results = restTemplate.getForObject(url, Singer.class);
    // Artificial delay of 1s for demonstration purposes
    Thread.sleep(1000L);
    return CompletableFuture.completedFuture(results);
}

@Async
public CompletableFuture<Writer> lookForId(String id) throws InterruptedException {
    LOG.info("Looking up Movie ID: {}", id);
    String url = "http://localhost:7078/data/" + id;
    Writer results = restTemplate.getForObject(url, Writer.class);
    // Artificial delay of 1s for demonstration purposes
    Thread.sleep(1000L);
    return CompletableFuture.completedFuture(results);
}

调用它们的函数就像下面的代码

public String getResp() throws InterruptedException, ExecutionException {
    long start = System.currentTimeMillis();
    CompletableFuture<Actor> page1 = actorAsyncService.lookForId("0");
    CompletableFuture<Singer> page2 = singerAsyncService.lookForId("2");
    CompletableFuture<Writer> page3 = singerAsyncService.lookForId("4");
    String respStr = page1.get() + "||" + page2.get() + "||" + page3.get() ;
    System.out.println(">>>>>>>>>>>> Elapsed time: " + (System.currentTimeMillis() - start));
    System.out.println("respStr : " + respStr);
    return respStr;
}

这里的进程等待每个进程完成，然后发回响应，但我希望一旦任何进程完成，那么服务应该将响应返回给调用函数但当进程的其余部分完成或终止或给出错误响应时，还应该能够处理其他服务调用提供的响应。

例如如果 page2 进程完成，则该进程的响应返回到调用函数，但该函数仍然应该能够在进程完成时处理来自 page1 和 page2 的响应。

Answer 1

CompletableFuture.anyOf() 返回一个新的 CompletableFuture，当任何给定的 CompletableFuture 完成时，该新的 CompletableFuture 就会完成，并具有相同的结果。

这是我的例子:

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task 1 finished");
            return "task 1";
        });

        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task 2 finished");
            return "task 2";
        });

        CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task 3 finished");
            return "task 3";
        });
        CompletableFuture<Object> future = CompletableFuture.anyOf(task1, task2, task3);
        String result = (String)future.get();
        System.out.println(result);

        task1.join();
        task2.join();
        task3.join();
    }

输出是:

task 1 finished
task 1
task 2 finished
task 3 finished