一、CompletableFuture用法入门介绍
入门介绍的一个例子：
package com.cy.java8; import java.util.Random;
import java.util.concurrent.CompletableFuture; public class CompletableFutureInAction {
private final static Random RANDOM = new Random(System.currentTimeMillis()); public static void main(String[] args){
CompletableFuture<Double> completableFuture = new CompletableFuture<>();
new Thread(() -> {
double value = get();
completableFuture.complete(value);
}).start(); System.out.println("do other things..."); completableFuture.whenComplete((t, e) -> {
System.out.println("complete. value = "+ t);
if(e != null){
e.printStackTrace();
}
});
} private static double get(){
try {
Thread.sleep(RANDOM.nextInt(3000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return RANDOM.nextDouble();
}
}
console打印：
do other things...
complete. value = 0.8244376567363494
二、CompletableFuture.supplyAsync
CompletableFuture很少有直接new出来的方式去用的，一般都是通过提供的静态方法来使用。
1.使用CompletableFuture.supplyAsync来构造CompletableFuture：
package com.cy.java8;
import java.util.concurrent.*;
import static com.cy.java8.CompletableFutureInAction.get;
public class CompletableFutureInAction2 {
public static void main(String[] args) {
/**
* 可以发现value=..没有被打印，为什么呢？
* 因为此方法构造出来的线程是demon的，守护进程，main执行结束之后就消失了，所以
* 根本没来得及执行whenComplete中的语句
*/
CompletableFuture.supplyAsync(() -> get())
.whenComplete((v, e) -> {
System.out.println("value = " + v);
if (e != null) {
e.printStackTrace();
}
});
System.out.println("do other things...");
}
}2.要将上面whenComplete中的语句执行，进行改造：
package com.cy.java8; import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import static com.cy.java8.CompletableFutureInAction.get; public class CompletableFutureInAction2 { public static void main(String[] args) throws InterruptedException {
AtomicBoolean finished = new AtomicBoolean(false); CompletableFuture.supplyAsync(() -> get())
.whenComplete((v, e) -> {
System.out.println("value = " + v);
if (e != null) {
e.printStackTrace();
}
finished.set(true);
}); System.out.println("do other things..."); while(!finished.get()){
Thread.sleep(1);
}
} }
改写之后， main线程发现如果finished没有变为true就会一直等1ms，直到whenComplete执行将finished变为true。
3.上面的改写很low，其实只要将守护线程变为前台进程，main结束后不会消失就行了。
package com.cy.java8; import java.util.concurrent.*;
import static com.cy.java8.CompletableFutureInAction.get; public class CompletableFutureInAction2 { public static void main(String[] args){
ExecutorService executorService = Executors.newFixedThreadPool(2, r -> {
Thread t = new Thread(r);
t.setDaemon(false); //非守护线程
return t;
}); CompletableFuture.supplyAsync(() -> get(), executorService)
.whenComplete((v, e) -> {
System.out.println("value = " + v);
if (e != null) {
e.printStackTrace();
}
}); System.out.println("do other things..."); //main执行结束之后，executorService线程不会结束，需要手动shutdown
executorService.shutdown();
} }
三、thenApply：
package com.cy.java8; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; public class CompletableFutureInAction3 { public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
Thread t = new Thread(r);
t.setDaemon(false);
return t;
}); /**
* 将执行完的结果再*100
*/
CompletableFuture.supplyAsync(CompletableFutureInAction::get, executor)
.thenApply(v -> multiply(v))
.whenComplete((v, e) -> System.out.println(v));
} private static double multiply(double value){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return value * 100;
} }
console打印：
43.15351824222534
四、CompletableFuture.join()
package com.cy.java8; import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors; public class CompletableFutureInAction3 { public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
Thread t = new Thread(r);
t.setDaemon(false);
return t;
}); /**
* 需求：将一组商品列表里面的每个商品对应的价格查询出来，并将这个价格*100.
* 5个商品同时并发去做这件事
*
* CompletableFuture.join():等到所有的结果都执行结束，会返回CompletableFuture自己本身
* 执行完的结果，等于get()返回的结果。
*/
List<Integer> productionIDs = Arrays.asList(1, 2, 3, 4, 5); //待查的一组商品列表的ID
List<Double> priceList = productionIDs.stream().map(id -> CompletableFuture.supplyAsync(() -> queryProduction(id), executor))
.map(future -> future.thenApply(price -> multiply(price)))
.map(multiplyFuture -> multiplyFuture.join())
.collect(Collectors.toList());
System.out.println(priceList); /**
* 按照以前，要5个分别for循环去查询
* 或者分多个线程去查询，再将每个线程查询的结果汇总，等到全部线程都执行完了，结果也就出来了
*/
} private static double multiply(double value) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return value * 100;
} /**
* 模拟 根据商品id查询对应的价格
* @param id
* @return
*/
private static double queryProduction(int id){
return CompletableFutureInAction.get();
}
}
console打印：
[90.93730009374265, 23.65282935900653, 17.415066430776815, 16.605197824452343, 60.143109082288206]
说明：这里多个任务同时执行，最终把结果汇总到一起 ，这种都是并行去执行的，编写代码也比较简洁，不需要考虑多线程之间的一些交互、锁、多线程之间的通信、控制，都不需要去关心。
五、CompletableFuture的常用API介绍
supplyAsync
thenApply
whenComplete
handle
thenRun
thenAccept
thenCompose
thenCombine
thenAcceptBoth
runAfterBoth
applyToEither
acceptEither
runAfterEither
anyOf
allOf
1）supplyAsync、thenApply、whenComplete前面的代码已经介绍了。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.thenApply(v -> Integer.sum(v,10))
.whenComplete((v, e) -> System.out.println(v));
Thread.sleep(1000);
}
}2）whenCompleteAsync: whenComplete是同步的方式，如果对于结果的处理是比较占时间的，不想通过这种同步的方式去做，可以用whenCompleteAsync进行异步操作。
3）handle：和thenApply差不多，只是多了一个对于异常的考虑。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.handle((v, e) -> Integer.sum(v, 10))
.whenComplete((v, e) -> System.out.println(v));
Thread.sleep(1000);
}
}4）thenRun：如果想在completableFuture整个执行结束之后，还想进行一个操作，可以thenRun(Runnable r)
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.handle((v, e) -> Integer.sum(v, 10))
.whenComplete((v, e) -> System.out.println(v))
.thenRun(()-> System.out.println("thenRunning..."));
Thread.sleep(1000);
}
}11
thenRunning...
5）thenAccept: thenAccept(Consumer c)里面传的是consumer，对执行结果进行消费，不会对执行结果进行任何操作。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.thenApply(v -> Integer.sum(v, 10))
.thenAccept(System.out::println);
Thread.sleep(1000);
}
}11
6）thenCompose： 对执行结果再交给另外一个CompletableFuture，它再去对这个执行结果进行另外的计算。compose：组合，组合设计模式。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.thenCompose(v -> CompletableFuture.supplyAsync(() -> v * 10))
.thenAccept(System.out::println);
Thread.sleep(1000);
}
}10
7）thenCombine： thenCombine(CompletableFuture extends CompletionStage, BiFuntion)
CompletableFuture的计算结果v1作为BiFunction的第1个入参，thenCombine中的第一个参数CompletableFuture的计算结果v2作为BiFunction的第2个入参，biFunction进行操作然后返回结果。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.thenCombine(CompletableFuture.supplyAsync(() -> 2.0), (v1, v2) -> v1 + v2)
.thenAccept(System.out::println);
Thread.sleep(1000);
}
}3.0
8）thenAcceptBoth： 和thenCombine差不多，只不过它的第二个参数是BiConsumer
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction4 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> 1)
.thenAcceptBoth(CompletableFuture.supplyAsync(() -> 2.0), (v1, v2) -> {
System.out.println("value=" + (v1 + v2));
});
Thread.sleep(1000);
}
}value=3.0
9）runAfterBoth：两个CompletableFuture都执行结束之后，run
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction5 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
return 1;
}).runAfterBoth(CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " is running too");
return 2;
}), () -> System.out.println("both done"));
Thread.sleep(1000);
}
}ForkJoinPool.commonPool-worker-9 is running
ForkJoinPool.commonPool-worker-9 is running too
both done
10）applyToEither
applyToEither：两个CompletableFuture只要有1个执行完了，就将这个CompletableFuture交给Function。谁先执行完就将谁交给Function去执行
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction5 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(900);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 1");
return 1;
}).applyToEither(CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 2");
return 2;
}), v -> {
System.out.println("value = " + v);
return v * 10;
}).thenAccept(System.out::println);
Thread.currentThread().join();
}
}I am future 2
value = 2
20
I am future 1
11）acceptEither
acceptEither：acceptEither(CompletableFuture extends CompletionStage, Consumer), 两个CompletableFuture谁先执行完成，就将谁的结果交给consumer执行。
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction5 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(900);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 1");
return 1;
}).acceptEither(CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 2");
return 2;
}), v -> System.out.println("value = " + v));
Thread.currentThread().join();
}
}I am future 2
value = 2
I am future 1
12）runAfterEither
runAfterEither: runAfterEither(CompletionStage, Runnable),只要有一个CompletableFuture执行完了，就执行run
package com.cy.java8;
import java.util.concurrent.CompletableFuture;
public class CompletableFutureInAction5 {
public static void main(String[] args) throws InterruptedException {
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(900);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 1");
return 1;
}).runAfterEither(CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("I am future 2");
return 2;
}), () -> System.out.println("done."));
Thread.currentThread().join();
}
}I am future 2
done.
I am future 1
13）allOf
allOf(CompletableFuture<?>... cfs)，返回值是CompletableFuture<Void>。要等所有的CompletableFuture都执行完成，才能执行下一步动作。
package com.cy.java8; import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors; public class CompletableFutureInAction5 {
private final static Random RANDOM = new Random(System.currentTimeMillis()); public static void main(String[] args) throws InterruptedException {
List<CompletableFuture<Double>> list = Arrays.asList(1, 2, 3, 4).stream()
.map(i -> CompletableFuture.supplyAsync(CompletableFutureInAction5::get))
.collect(Collectors.toList()); //要等所有的CompletableFuture这些task执行完了，才会打印done.
CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()]))
.thenRun(() -> System.out.println("done.")); Thread.currentThread().join();
} static double get() {
try {
Thread.sleep(RANDOM.nextInt(3000));
} catch (InterruptedException e) {
e.printStackTrace();
}
double result = RANDOM.nextDouble();
System.out.println(result);
return result;
}
}
0.6446554001163166
0.24435437709196395
0.18251850071600362
0.5261702037394511
done.
14）anyOf
和allOf相反，只要有一个CompletableFuture执行完成，就会执行下一步动作
package com.cy.java8; import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors; public class CompletableFutureInAction5 {
private final static Random RANDOM = new Random(System.currentTimeMillis()); public static void main(String[] args) throws InterruptedException {
List<CompletableFuture<Double>> list = Arrays.asList(1, 2, 3, 4).stream()
.map(i -> CompletableFuture.supplyAsync(CompletableFutureInAction5::get))
.collect(Collectors.toList()); //只要有一个CompletableFuture执行完了，就会打印done.
CompletableFuture.anyOf(list.toArray(new CompletableFuture[list.size()]))
.thenRun(() -> System.out.println("done.")); Thread.currentThread().join();
} static double get() {
try {
Thread.sleep(RANDOM.nextInt(3000));
} catch (InterruptedException e) {
e.printStackTrace();
}
double result = RANDOM.nextDouble();
System.out.println(result);
return result;
}
}
0.1334361442807943
done.
0.6715112881360222
0.12945359790698785
0.1307762755130788
----