Future接口定义了异步任务执行的一些方法,包括异步任务执行结果,异步任务执行是否中断,异步任务是否完毕等。
FutureTask<String> futureTask = new FutureTask<String>( () -> {
System.out.println(Thread.currentThread().getName()+"\t -----come in");
try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
return "task over";
});
Thread t1 = new Thread(futureTask, "t1");
t1.start();
stage.thenApply(x -> square(x)).thenAccept(×->System.out.print(x)).thenRun(( ->system.out.println())
主要是runAsync和supplyAsnc方法。一个无返回值。一个有返回值。
join和get的区别。get必须处理异常。join不需要处理异常
jdk8新特性: lambda表达式,stream流,chain链式调用,函数式编程
有参数,有返回值:Function
有参数,无返回值:Consume, BiConsumer(两个参数)
无参数,有返回值:Supplier
无参数,无返回值:Runnable
原来的写法,串行的方式:
/**
* step by step 一家家搜查
* List<NetMall> ----->map------> List<String>
* @param list
* @param productName
* @return
*/
public static List<String> getPrice(List<NetMall> list,String productName)
{
//《mysql》 in taobao price is 90.43
return list
.stream()
.map(netMall ->
String.format(productName + " in %s price is %.2f",
netMall.getNetMallName(),
netMall.calcPrice(productName)))
.collect(Collectors.toList());
}
使用CompletableFuture,异步的方式:
/**
* List<NetMall> ----->List<CompletableFuture<String>>------> List<String>
* @param list
* @param productName
* @return
*/
public static List<String> getPriceByCompletableFuture(List<NetMall> list,String productName)
{
return list.stream().map(netMall ->
CompletableFuture.supplyAsync(() -> String.format(productName + " in %s price is %.2f",
netMall.getNetMallName(),
netMall.calcPrice(productName))))
.collect(Collectors.toList())
.stream()
.map(s -> s.join())
.collect(Collectors.toList());
}
耗时:比串行的方式快得多!!!
获得结果:
主动触发计算:
public class CompletableFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "hello CompletableFuture";
});
System.out.println(completableFuture.getNow("心急吃不了热豆腐"));
System.out.println(completableFuture.get());
System.out.println(completableFuture.get(1500, TimeUnit.MILLISECONDS));
System.out.println(completableFuture.join());
System.out.println(completableFuture.complete("未雨绸缪")+"\t"+completableFuture.join());
}
}
两个计算结果存在依赖关系,这两个线程串行化。
出现异常,直接跳到whenComplete和exceptionally执行。(不再执行后续的thenApply)
public class CompletableFutureTest2 {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 6;
},executorService).thenApply((r)-> {
int i=2/0;
return r * 5;
}).thenApply((r)-> {
System.out.println(r);
return r - 2;
}).whenComplete((v, e) -> {
System.out.println("计算结果:"+v);
}).exceptionally(e -> {
System.out.println(e.getMessage());
System.out.println(e);
return null;
});
System.out.println("============主线程==========");
executorService.shutdown();
}
}
计算机结果存在依赖关系,两个线程串行化
handle出现异常,会往下一个handle走,同时也会走到whenComplete和exceptionally
public class CompletableFutureTest2 {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 6;
},executorService).handle((r,e)-> {
int i=2/0;
return r * 5;
}).handle((r,e)-> {
System.out.println(r);
return r - 2;
}).whenComplete((v, e) -> {
System.out.println("计算结果:"+v);
}).exceptionally(e -> {
System.out.println(e.getMessage());
System.out.println(e);
return null;
});
System.out.println("============主线程==========");
executorService.shutdown();
}
}
exceptionally相当于try catch
whenComplete和handler相当于try finally
接受任务的处理结果,消费处理。thenAccept无返回结果。(thenApply是有返回结果的)
public class CompletableFutureTest3 {
public static void main(String[] args) {
CompletableFuture.supplyAsync(()->{
return 3;
}).thenApply(r->{
return r*8;
}).thenApply(r->{
return r/2;
}).thenAccept(r-> System.out.println(r));
System.out.println(CompletableFuture.supplyAsync(()->"6666").thenRun(()->{}).join());
System.out.println(CompletableFuture.supplyAsync(()->"6666").thenAccept(r-> System.out.println(r)).join());
System.out.println(CompletableFuture.supplyAsync(()->"6666").thenApply(r->r+"9999").join());
}
}
12
null
6666
null
66669999
applyToEither:谁快用谁
thenCombine: 两个completionStage任务都完成后,将结果交给thenCombine。先完成的先等着,等待其他分支任务。
public class CompletableFutureTest4 {
public static void main(String[] args) {
CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "1号选手";
});
CompletableFuture<String> second = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "2号选手";
});
CompletableFuture<String> result = first.applyToEither(second, r -> r + "is winner");
CompletableFuture<String> res = first.thenCombine(second, (x, y) -> x + y);
System.out.println(result.join());
System.out.println(res.join());
}
}
allOf():当所有给定的CompletableFuture完成时, 返回一个新的CompletableFuture
anyOf():当任何一个给定的CompletableFuture完成时,返回一个新的CompletableFuture
public static void testAllOf(){
CompletableFuture<String> future1 = CompletableFuture.runAsync(() -> {
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("future1执行完成");
});
CompletableFuture<String> future2 = CompletableFuture.runAsync(() -> {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("future2执行完成");
});
CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2);
try {
all.get(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
以thenRun和thenRunAsync为例,有什么区别?