CompletableFuture中的CompletionException异常真是坑到我了!

  • Post author:
  • Post category:其他


开发环境

  • JDK 17

  • Idea 2022

熟悉JDK 8版本的同学,大概率都使用过

java.util.concurrent.CompletableFuture

这个类,有时候在业务服务中你可能需要并行的去执行某些骚操作,那就少不了它的存在。

我在业务中就有大量的需求场景,比如:我需要拉渠道方的订单和订单明细,落地到我司系统中。此时用到了两个接口:订单列表查询、订单明细查询,查询列表很简单,一页一页的翻页查询,因为它也只返回了订单号,我举个例子吧!

这里是订单列表查询接口返回的数据结构(Mock数据),一堆的订单号

{
    "data":[
        "125345345345",
        "235894563423",
        "345345345343"
    ]
}

接下来我肯定要根据订单号去查询订单明细的,千万不要下意识的进行for循环一个一个查,那要查到什么时候啊!

这个时候

CompletableFuture

就派上用场了。

CompletableFuture应用

前提条件,我为了方便使用,我是直接在业务的Base包中,封装了一个通用实现,当然只是适用于我们业务的,需要的同学自取。

注意:下面这段代码没有处理真实的异常,想直接用,再往下看!!!


/**
 * 偷个懒,线程池直接这样写了先,真实业务中不是这样搞的哈!
 */
private final static ExecutorService executorService = Executors.newFixedThreadPool(4);


/**
 * 创建并行任务并执行
 *
 * @param list            数据源
 * @param api             API调用逻辑
 * @param exceptionHandle 异常处理逻辑
 * @param <S>             数据源类型
 * @param <T>             程序返回类型
 * @return 处理结果列表
 */
public <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> api, BiFunction<Throwable, S, T> exceptionHandle) {
    //规整所有任务
    List<CompletableFuture<T>> collectFuture = list.stream()
            .map(s -> this.createFuture(() -> api.apply(s), e -> exceptionHandle.apply(e, s)))
            .toList();
    //汇总所有任务,并执行join,全部执行完成后,统一返回
    return CompletableFuture.allOf(collectFuture.toArray(new CompletableFuture<?>[]{}))
            .thenApply(f -> collectFuture.stream()
                    .map(CompletableFuture::join)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList()))
            .join();
}

/**
 * 创建单个CompletableFuture任务
 *
 * @param logic           任务逻辑
 * @param exceptionHandle 异常处理
 * @param <T>             类型
 * @return 任务
 */
public <T> CompletableFuture<T> createFuture(Supplier<T> logic, Function<Throwable, T> exceptionHandle) {
    return CompletableFuture.supplyAsync(logic, executorService).exceptionally(exceptionHandle);
}

利用上面这块封装的代码,完全适用于我在公司的大部分并行业务场景,也确实提升了我Pod节点的CPU利用率。

不过之后有个问题就坑了,你调用外部API的时候,偶尔会失败,那么失败就要重试,这个时候我就需要正确的判断异常并进行重试操作。

先定义个业务异常类

public static class BizApiException extends RuntimeException {
    public BizApiException() {
    }

    public BizApiException(String message) {
        super(message);
    }
}

示例代码

下面的代码只是模拟我在业务端的场景,大家乐呵乐呵就行。

public static void main(String[] args) {
    CompletableFutureDemo f = new CompletableFutureDemo();
    List<Integer> numList = f.parallelFutureJoin(Arrays.asList(1, 2, 3), num -> {
        //模拟API调用
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            //...
        }
        if (num > 2) {
            throw new BizApiException("心别太大");
        }
        return num;
    }, (e, num) -> {
        //异常向你打开了大门
        if (e instanceof BizApiException) {
            System.out.println("业务异常,我在处理数字:" + num + ",异常原因:" + e);
            return -1;
        }
        System.out.println("我异常了,老六,我刚才在处理数字:" + num + ",异常原因:" + e);
        return -1;
    });
    System.out.println(numList);
}

注意:我本来想通过返回的Exception判断是不是BizApiException业务异常的,可惜的是这里的异常类型永远都不会是BizApiException,我输出下内容到控制台,就懂了!

执行代码后,我得到的控制台内容

这个时候其实可以看到拿到的真实异常类型了,一个名为

java.util.concurrent.CompletionException

的老六出现了。我看看它是个啥!

我异常了,老六,我刚才在处理数字:3,异常原因:java.util.concurrent.CompletionException: com.java.basic.CompletableFutureDemo$BizApiException: 心别太大
[1, 2, -1]

CompletionException

这个类的注释说明了它的用途,简单理解是:它在完成结果或任务的过程中遇到错误或其他异常时会触发。那我懂了!任务异常之后它就会出现,然后还把我们自己的异常给包起来了!

JDK源代码

/**
 * Exception thrown when an error or other exception is encountered
 * in the course of completing a result or task.
 *
 * @since 1.8
 * @author Doug Lea
 */
public class CompletionException extends RuntimeException {}

其实还有个异常类,我们也需要关注,它告诉了我们,真实的异常在哪里了!

ExecutionException

这个类的注释就说的很清楚了,自己的异常都在getCause()中了,那就好办了.

/**
 * Exception thrown when attempting to retrieve the result of a task
 * that aborted by throwing an exception. This exception can be
 * inspected using the {@link #getCause()} method.
 *
 * @see Future
 * @since 1.5
 * @author Doug Lea
 */
public class ExecutionException extends Exception {}

改造我的并发工具类(完整版)

方法extractRealException就是我要获取的真实异常,同时parallelFutureJoin方法中引用一下,这个工具类就解决了需求。

/**
 * 创建并行任务并执行
 *
 * @param list            数据源
 * @param api             API调用逻辑
 * @param exceptionHandle 异常处理逻辑
 * @param <S>             数据源类型
 * @param <T>             程序返回类型
 * @return 处理结果列表
 */
public <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> api, BiFunction<Throwable, S, T> exceptionHandle) {
    //规整所有任务
    List<CompletableFuture<T>> collectFuture = list.stream()
            .map(s -> this.createFuture(() -> api.apply(s), e -> exceptionHandle.apply(
this.extractRealException(e), s)))
            .toList();
    //汇总所有任务,并执行join,全部执行完成后,统一返回
    return CompletableFuture.allOf(collectFuture.toArray(new CompletableFuture<?>[]{}))
            .thenApply(f -> collectFuture.stream()
                    .map(CompletableFuture::join)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList()))
            .join();
}

/**
 * 创建CompletableFuture任务
 *
 * @param logic           任务逻辑
 * @param exceptionHandle 异常处理
 * @param <T>             类型
 * @return 任务
 */
public <T> CompletableFuture<T> createFuture(Supplier<T> logic, Function<Throwable, T> exceptionHandle) {
    return CompletableFuture.supplyAsync(logic, executorService).exceptionally(exceptionHandle);
}

/**
 * 提取真正的异常
 * <p>
 * CompletableFuture抛出的往往不是真正的异常
 *
 * @param throwable 异常
 * @return 真正的异常
 */
public Throwable extractRealException(Throwable throwable) {
    if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
        if (throwable.getCause() != null) {
            return throwable.getCause();
        }
    }
    return throwable;
}

当然,这还只是个简易版的并行任务工具类,还有更多的可能,大家需要自己去探索了!

参考文献



版权声明:本文为u010185035原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。