Spring Boot+Java中的异步调用

Question

我有一个场景，我按顺序进行大量验证，例如..

public class validationClass{
  public validate(ValidationObject validationObj){
   validation1(validationObj);
   validation2(validationObj);
   validation3(validationObj);
   validation4(validationObj);
   ...
   validationN(validationObj);
   return validationObj;
  }
}

validationObject 被传递到每个验证中，如果存在任何错误，都将附加到其中，最后将返回对象。 目前，由于执行是顺序的，因此需要更多时间。

我可以并行执行验证并合并每个线程中的响应并发回吗？

是否可以通过 Completable Futures 实现这一点？我找不到这方面的例子。 请问哪位大神可以指点一下吗？

Answer 1

是的，可以使用CompletableFuture库来执行。

我可以与您分享一个示例，尝试为您的问题陈述实施相同的示例。

1. 创建将接受的 CompletableFuture 对象列表 ValidationObject 类型作为输入。

   List<CompletableFuture<ValidationObject>> completableFutureList = new ArrayList<>();
   

2. 将验证函数添加到列表中。

   completableFutureList.add(validation1(validationObj));
   completableFutureList.add(validation2(validationObj));
   

3. 验证方法必须返回相同的类型（在我的示例中：CompletableFuture）。 CompletableFuture 不接受 void 类型，并且至少必须返回某种类型。如果你的方法没有返回任何东西，那么至少返回一些字符串，例如：CompletableFuture。这里的“asyncExecutor”只不过是我们在 ThreadPoolTask​​Executor 配置中分配的自定义 @bean 名称。

此链接中的更多详细信息：https://www.oodlestechnologies.com/blogs/threadpooltaskexecutor-configuration-with-spring-boot/

    @Async("asyncExecutor")
    public CompletableFuture<String> validation1(ValidationObject validation)
    {
    //your code here
    }

4. 然后将各个异步任务连接在一起，这样执行程序只有在收到所有线程的结果后才能继续执行其他部分代码。如果您不加入，则执行其他代码，并且每个线程中的验证方法异步执行。

   for (CompletableFuture<ValidationObject> completableFuture : completableFutureList) 
   {
   completableFuture.join();
   }
   

Answer 2

您可以选择 java8 Streams's parallel processing。为了模拟您的场景，我创建了一些随机验证函数，对数据进行某种处理。

import java.util.stream.Stream;

public class StackOverFlow_65531407 {

  public static void main(String[] args) {

    StackOverFlow_65531407 stack = new StackOverFlow_65531407();

    ValidationObject vObj = new ValidationObject(1l, "name", "Password", "role");

    Long t1 = System.currentTimeMillis();
    stack.validate(vObj);
    Long t2 = System.currentTimeMillis();
    stack.validate_parallel(vObj);
    Long t3 = System.currentTimeMillis();

    System.out.println("---------------------------------------");
    System.out.println("time in non-parallel execution::" + (t2 - t1) + " milliseconds");
    System.out.println("time in parallel execution::" + (t3 - t2) + " milliseconds");
    System.out.println("---------------------------------------");

  }

  public ValidationObject validate(ValidationObject validationObj) {
    v1(validationObj);
    v2(validationObj);
    v3(validationObj);
    v4(validationObj);

    return validationObj;
  }

  public ValidationObject validate_parallel(ValidationObject validationObj) {

    Stream.of(validationObj)
        .parallel()
        .map(this::v1)
        .map(this::v2)
        .map(this::v3)
        .map(this::v4);

    return validationObj;
  }

  private ValidationObject v1(ValidationObject validationObj) {
    //do validation on id
    if (validationObj.getId() != null) {
      System.out.println(validationObj.getId());
      for (int i = 0; i < 10000; i++) {
        System.out.println("\tPrinted " + i);
      }
    }
    return validationObj;
  }

  private ValidationObject v2(ValidationObject validationObj) {
    //do validation on name
    if (validationObj.getName() != null) {
      System.out.println(validationObj.getName());
      for (int i = 0; i < 10000; i++) {
        System.out.println("\tPrinted " + i);
      }
    }
    return validationObj;
  }

  private ValidationObject v3(ValidationObject validationObj) {
    //do validation on password
    if (validationObj.getPassword() != null) {
      System.out.println(validationObj.getPassword());
      for (int i = 0; i < 10000; i++) {
        System.out.println("\tPrinted " + i);
      }
    }
    return validationObj;
  }

  private ValidationObject v4(ValidationObject validationObj) {
    //do validation on role
    if (validationObj.getRole() != null) {
      System.out.println(validationObj.getPassword());
      for (int i = 0; i < 1000; i++) {
        System.out.println("\tPrinted " + i);
      }
    }
    return validationObj;
  }

}

对于较少数量的输入（例如打印到 100 或数千），流并行会很慢。但在更大的图景中，stream out 运行顺序处理。

我的输出为：

----
---
    Printed 987
    Printed 988
    Printed 989
    Printed 990
    Printed 991
    Printed 992
    Printed 993
    Printed 994
    Printed 995
    Printed 996
    Printed 997
    Printed 998
    Printed 999
---------------------------------------
time in non-parallel execution::249 milliseconds
time in parallel execution::96 milliseconds
---------------------------------------

您可以看到并行流如何显着减少您的时间。但是使用并行流也是一种自以为是的选择。