CompletableFuture 使用示例

作者: AlienPaul | 来源:发表于2020-06-19 10:14 被阅读0次

背景

Java异步编程离不开Future接口，但是Future接口提供的方法使用起来不够灵活。为了判断一个Future是否已经完成，我们可以：

通过调用get方法，以阻塞的形式获取执行结果。
调用有最大等待时间的get方法，以阻塞的形式获取执行结果。
反复轮询isDone方法，直到任务完成，获取结果。

以上3种方法都不够灵活，会造成线程阻塞或耗费CPU资源。需要用户过多的参与异步编程逻辑，对业务代码的侵入性较强。

另外用户如果需要异步执行多个任务，并且这些任务具有先后依赖关系。基于传统的方式我们需要大量使用锁，CountDownLatch和CyclicBarrier和阻塞队列等，编程十分复杂。

CompletableFuture是Future的增强版，提供了一系列的同步或异步任务执行操作。除此之外还能够对异步任务多个阶段的前后依赖关系进行控制。使用起来十分方便。

注意事项

所有的以async结尾的方法都为异步执行。它们都可以传入一个Executor线程池，用于异步执行。如果不指定线程池，默认使用ForkJoinPool.commonPool()

下面以所有的async异步方法为例，说明下CompletableFuture的使用方法。

静态方法

supplyAsync

提供初始数据。接收一个Supplier类型参数。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> 1);
completableFuture.thenApplyAsync((i) -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return i + 1;
    // 这里打印出2
}).thenAcceptAsync((i) -> System.out.println(i));

runAsync

异步执行任务，接收Runnable类型参数。

CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("runAsync");
});

completedFuture

直接返回一个已经完成的CompletableFuture。接收一个任意类型数据为参数。

CompletableFuture<Integer> completableFuture = CompletableFuture.completedFuture(500);
// 此处可以立刻返回500
System.out.println(completableFuture.get());

allOf

参数接收任意多个CompletableFuture。该方法返回一个新的CompletableFuture，只有在参数所有的CompletableFuture完成的时候它才会完成。

long start = System.currentTimeMillis();

CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("runAsync1");
});

CompletableFuture<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("runAsync2");
});

// 等待两个CompletableFuture都完成的时候，future才会完成
CompletableFuture<Void> future = CompletableFuture.allOf(completableFuture1, completableFuture2);

// 阻塞到CompletableFuture1完成
future.get();
// 耗时大于3000毫秒
System.out.println(System.currentTimeMillis() - start);

anyOf

和allOf一样，返回一个新的CompletableFuture，参数中任意一个CompletableFuture完成时它就能完成。

long start = System.currentTimeMillis();

CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("runAsync1");
});

CompletableFuture<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("runAsync2");
});

// 两个CompletableFuture有任何一个完成，future就可以完成
CompletableFuture<Object> future = CompletableFuture.anyOf(completableFuture1, completableFuture2);

// 阻塞到CompletableFuture2完成
future.get();
// 耗时略大于1000毫秒
System.out.println(System.currentTimeMillis() - start);

普通方法

thenApplyAsync

进行数据处理，接收前一步骤传递的数据，处理加工后返回。返回数据类型可以和前一步骤返回的数据类型不同。
接收参数为Function类型。

CompletableFuture.supplyAsync(() -> 1).thenApplyAsync((i) -> "Hi: " + i).whenCompleteAsync(((s, throwable) -> {
    // 返回 Hi: 1
    System.out.println(s);
}));

thenAcceptAsync

接收上游传递过来的数据并消费。接收一个Consumer类型参数。

CompletableFuture.supplyAsync(() -> 1).thenAcceptAsync(System.out::println);

acceptEitherAsync

接收另一个CompletableFuture和一个Consumer。含义为2个CompletableFuture哪个先运行完成，就采用谁的执行结果。

CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
}).acceptEitherAsync(completableFuture1, integer -> {
    // completableFuture1最早返回，所以integer的值为1
    System.out.println(integer);
});

这个例子中第一个CompletableFuture比第二个先执行完毕，因此acceptEitherAsync输出第一个CompletableFuture的结果。

applyToEitherAsync

接收一个CompletableFuture和一个Function。和acceptEitherAsync类似，只不过第二个参数从Consumer变成Function。

CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
}).applyToEitherAsync(completableFuture1, (integer -> {
    // 这里CompletableFuture1首先完成，所以integer为1
    return integer;
})).whenCompleteAsync((integer, throwable) -> {
    System.out.println(integer);
});

thenAcceptBothAsync

接收另一个CompletableFuture和BiConsumer，用于在两个CompletableFuture都执行完的时候，获取他们的执行结果并处理。

CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
}).thenAcceptBothAsync(completableFuture1, (integer, integer2) -> {
    // 等到两个CompletableFuture都完成之后回调
    // 在这个例子中需要等待5秒钟
    回调的两个参数分别对应两个CompletableFuture的执行结果
    System.out.println(integer + integer2);
});

runAfterEitherAsync

接收另一个CompletableFuture和Runnable类型参数。两个CompletableFuture有任意一个完成的时候，执行Runnable。

long start = System.currentTimeMillis();

CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
}).runAfterEitherAsync(completableFuture1, () -> {
    // 任何一个CompletableFuture完成的时候都会执行Runnable
    // 这个例子等待1秒后会执行Runnable，此处打印的值略大于1000
    System.out.println(System.currentTimeMillis() - start);
});

runAfterBothAsync

接收另一个CompletableFuture和Runnable类型参数。两个CompletableFuture都执行完成的时候，执行Runnable。

long start = System.currentTimeMillis();

CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
}).runAfterBothAsync(completableFuture1, () -> {
    // 和runAfterEitherAsync不同的是，这里必须等到两个CompletableFuture都完成的时候才会执行Runnable
    // 这里需要等待5秒钟后才会执行，打印的值略大于5000
    System.out.println(System.currentTimeMillis() - start);
});

complete

接收任意类型数据作为参数。如果CompletableFuture没有完成的时候调用complete()方法，后续再调用这个future的get()方法时返回这个值。

// 睡眠2秒再返回结果
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});
// 注意，此处如果睡眠3秒，调用complete时completableFuture已经执行完毕返回1，complete方法不会修改返回值。此时调用get方法返回1
// 如果此处没有睡眠3秒，调用complete时completableFuture尚未执行完毕，下面调用get的时候方法返回1000
Thread.sleep(3000);
completableFuture.complete(1000);

Integer integer = completableFuture.get();

System.out.println(integer);

completeExceptionally

接收一个Throwable类型参数。如果CompletableFuture没有完成的时候调用completeExceptionally()方法，后续再调用这个future的get()方法时会抛出这个异常。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

// 如果调用completeExceptionally的时候completableFuture没有执行完成，那么下面调用get的时候会抛出IllegalArgumentException
Thread.sleep(3000);
completableFuture.completeExceptionally(new IllegalArgumentException());

Integer integer = completableFuture.get();

System.out.println(integer);

cancel

相当于completeExceptionally(new CancellationException())，不再赘述。接收的参数对行为没有影响。

如果CompletableFuture没有执行完成的时候调用了cancel，cancel方法返回true。

如果CompletableFuture执行完成的时候调用了cancel，cancel方法返回false。

exceptionally

设置当CompletableFuture执行抛出异常时候的返回值。用于处理异常情况。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {

    // 抛出一个异常
    if (true) {
        throw new RuntimeException("Error Message");
    }

    return 1;
});

// 如果遇到异常，返回500
completableFuture.exceptionally((t) -> 500)
        .whenCompleteAsync(((integer, throwable) -> {
            // 打印出500
            System.out.println(integer);
            // 异常已被处理，返回null
            System.out.println(throwable.getMessage());
        }));

handleAsync

相比exceptionally更为复杂的处理exception方式。接收一个BiFunction类型的参数。

handleAsync也可以视为回调方法具有返回值的whenCompleteAsync。

注意和exceptionally的区别，handleAsync没有自动发现是否抛出exception的能力，需要手工编写相关逻辑。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {

    if (true) {
        throw new RuntimeException("Error Message");
    }

    return 1;
});


completableFuture
        .handleAsync((integer, throwable) -> {
            System.out.println(integer);
            System.out.println(throwable);
            return null == throwable ? 1 : 500;
        })
        .whenCompleteAsync(((integer, throwable) -> {
            System.out.println(integer);
            System.out.println(throwable.getMessage());
        }));

whenCompleteAsync

完成时异步回调，接收一个BiConsumer<? super T, ? super Throwable>类型参数。

例子在前面的代码中已有体现，不再赘述。

thenCombineAsync

含义相当于"thenApplyBothAsync"，但是JDK不叫这个名字。

接收另一个CompletableFuture和BiFunction作为参数。目的为等待两个CompletableFuture都完成的时候，执行BiFunction对两个CompletableFuture的返回值进行处理。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    return 1;
});

CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 10;
});

completableFuture.thenCombineAsync(completableFuture2, (integer, integer2) -> {
    // 等两个CompletableFuture都执行完毕后，返回他们两个返回值的和
    return integer + integer2;
}).whenCompleteAsync(((integer, throwable) -> {
    System.out.println(integer);
}));

thenComposeAsync

将两个CompletableFuture组合起来。接收的参数类型为Function。

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
    return 1;
});

completableFuture.thenComposeAsync(integer -> {
    // 这里返回另一个CompletableFuture，该CompletableFuture可以使用前面CompletableFuture的返回值进行计算
    return CompletableFuture.supplyAsync(() -> {
        return 10 + integer;
    });
}).whenCompleteAsync(((integer, throwable) -> {
    // 返回11
    System.out.println(integer);
}));