CompletableFuture异步回调

CompletableFuture异步回调

CompletableFuture简介

CompletableFuture被用于异步编程，异步通常意味着非阻塞，可以使得任务单独允许在与主线程分离的其他线程中，并且通过回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常信息。
CompletableFuture实现了Future，CompletionStage接口，实现了Future接口可以兼容线程池框架，而CompletionStage接口才是异步编程的接口抽象，里面定义多种异步方法，通过这两者集合，从而打造出了强大的CompletableFuture类。
Futrue和CompletableFuture

Future在Java里面，通过用来表示一个异步任务的引用，比如我们将任务提交到线程池里面，然后我么会得到一个Future，在Future里面有isDone方法来判断任务是否处理结束，该有get方法可以一直阻塞直到任务结束然后获取结果，但整体来说这种方式，还是同步的，因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。
Futrue缺点

1.不支持手动完成。2.不支持进一步的非阻塞调用。3.不支持链式调用。4.不支持多个Future合并。5.不支持异步处理。
CompletableFuture类的使用案例

CompletableFuture01

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
6. /**
7. * 演示CompletableFuture
8. * @author 长名06
9. * @version 1.0
10. */
11. public class CompletableFuture01 {
12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
13.         CompletableFuture<String> future = new CompletableFuture<>();
15.         new Thread(() -> {
16.             System.out.println("子线程开始干活");
17.             try {
18.                 //子线程沉睡3s
19.                 Thread.sleep(3000);
20.             } catch (InterruptedException e) {
21.                 e.printStackTrace();
22.             }
23.             //完成future任务
24.             future.complete("success");
25.         },"A").start();
27.         System.out.println("主线程调用get方法获取结果为:" + future.get());
28.         System.out.println("主线程完成，阻塞结束");
29.     }
30. }

复制代码

CompletableFuture02

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
6. /**
7. * @author 长名06
8. * @version 1.0
9. */
10. public class CompletableFuture02 {
11.     public static void main(String[] args) throws ExecutionException, InterruptedException {
13.         //异步调用，无返回值
14.         CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
15.             try {
16.                 Thread.sleep(3000);
17.             } catch (InterruptedException e) {
18.                 e.printStackTrace();
19.             }
20.             System.out.println(Thread.currentThread().getName() + "执行runSync()");
21.         });
22.         completableFuture1.get();
24.         //异步调用，有返回值
25.         CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
26.             System.out.println(Thread.currentThread().getName() + "执行supplyAsync()");
27. //            int i = 1/0;
28.             return 1024;
29.         });
30.         completableFuture2.whenComplete((t, u) -> {
31.             System.out.println("----t=" + t);//t参数，是执行的返回值
32.             System.out.println("----u=" + u);//异常信息
33.         }).get();
34. //        System.out.println(Runtime.getRuntime().availableProcessors());
36.     }
37. }

复制代码

CompletableFuture03

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
6. /**
7. * 演示线程依赖，执行api thenApply()
8. * 一个任务，依赖于另一个任务可以使用thenApply()将两个任务(线程)串行化
9. * 对一个数先加10 再平方
10. * @author 长名06
11. * @version 1.0
12. */
13. public class CompletableFuture03 {
15.     public static Integer num = 10;
17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
18.         System.out.println(Thread.currentThread().getName() + "主线程开始");
21.         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
22.             System.out.println("加10任务开启");
23.             num += 10;
24.             return num;
25.         }).thenApply(i -> num * num);
26.         Integer integer = future.get();
27.         System.out.println("主线程结束，子线程的结果为" + integer);
28.     }
29. }

复制代码

CompletableFuture04

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
5. import java.util.function.Consumer;
6. import java.util.function.Function;
8. /**
9. * 消费处理结果
10. * thenAccept()方法，接收任务的处理结果，并消费结果，不返回结果了
11. * @author 长名06
12. * @version 1.0
13. */
14. public class CompletableFuture04 {
15.     public static Integer num = 10;
17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
18.         System.out.println(Thread.currentThread().getName() + "主线程开始");
21.         CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
22.             System.out.println("加10任务开启");
23.             num += 10;
24.             return num;
25.         }).thenApply(new Function<Integer, Integer>() {
26.             @Override
27.             public Integer apply(Integer integer) {
28.                 return num * num;
29.             }
30.         }).thenAccept(new Consumer<Integer>() {
31.             @Override
32.             public void accept(Integer i) {
33.                 System.out.println("子线程全部处理完成，最后调用了accept方法，消费了结果" + i);
34.             }
35.         });
36.     }
37. }

复制代码

CompletableFuture05

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
5. import java.util.function.Consumer;
6. import java.util.function.Function;
8. /**
9. * 异常处理
10. * @author 长名06
11. * @version 1.0
12. */
13. public class CompletableFuture05 {
14.     public static Integer num = 10;
16.     public static void main(String[] args) throws ExecutionException, InterruptedException {
17.         System.out.println(Thread.currentThread().getName() + "主线程开始");
20.         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
21.             int i = 1/0;//模拟异常
22.             System.out.println("加10任务开启");
23.             num += 10;
24.             return num;
25.         }).exceptionally(new Function<Throwable, Integer>() {
26.             @Override
27.             public Integer apply(Throwable ex) {
28.                 System.out.println(ex.getMessage());
29.                 return -1;
30.             }
31.         });
32.     }
33. }

复制代码

CompletableFuture06

1. package com.shaonian.juc.completable;
3. import java.util.concurrent.CompletableFuture;
4. import java.util.concurrent.ExecutionException;
5. import java.util.function.BiFunction;
6. import java.util.function.Function;
8. /**
9. * 消费结果，同时处理异常
10. * handle类似与thenAccept/thenRun方法，是最后一步结果的调用，但是同时可以处理异常
11. * @author 长名06
12. * @version 1.0
13. */
14. public class CompletableFuture06 {
15.     public static Integer num = 10;
17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
18.         System.out.println(Thread.currentThread().getName() + "主线程开始");
21.         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
22.                 int i = 1/0;
23.                 System.out.println("加10任务开启");
24.                 num += 10;
25.             return num;
26.         }).handle(new BiFunction<Integer, Throwable, Integer>() {
27.             @Override
28.             public Integer apply(Integer i, Throwable ex) {
29.                 System.out.println("进入了handle方法");
30.                 if(ex != null){
31.                     System.out.println("发生了异常,内容为" + ex.getMessage());
32.                     return -1;
33.                 }else{
34.                     System.out.println("正常执行,结果为" + i);
35.                     return i;
36.                 }
37.             }
38.         });
39.     }
40. }

复制代码

CompletableFuture07

1. package com.shaonian.juc.completable;
3. import java.util.ArrayList;
4. import java.util.List;
5. import java.util.concurrent.CompletableFuture;
6. import java.util.concurrent.CompletionStage;
7. import java.util.concurrent.ExecutionException;
8. import java.util.function.BiFunction;
9. import java.util.function.Function;
11. /**
12. * 两个CompletableFuture结果的合并
13. * @author 长名06
14. * @version 1.0
15. */
16. public class CompletableFuture07 {
17.     public static Integer num = 10;
19.     public static void main(String[] args) throws ExecutionException, InterruptedException {
21.         //有依赖关系的合并
22.         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
23.             System.out.println("加10任务开启");
24.             num += 10;
25.             return num;
26.         });
27.         //合并
28.         CompletableFuture<Integer> future2 = future.thenCompose(new Function<Integer, CompletionStage<Integer>>() {
29.             @Override
30.             public CompletionStage<Integer> apply(Integer i) {
31.                 return CompletableFuture.supplyAsync(() -> {
32.                     return i + 1;
33.                 });
34.             }
35.         });
36.         System.out.println(future.get());
37.         System.out.println(future2.get());
39.         //无依赖的任务合并
40.         CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
41.             System.out.println("加10任务开启");
42.             num += 10;
43.             return num;
44.         });
46.         CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
47.             System.out.println("乘10任务开启");
48.             num *= 10;
49.             return num;
50.         });
52.         //合并两个结果
53.         CompletableFuture<Object> future3 = job1.thenCombine(job2, new BiFunction<Integer, Integer, List<Integer>>() {
54.             @Override
55.             public List<Integer> apply(Integer result1, Integer result2) {
56.                 ArrayList<Integer> list = new ArrayList<>();
57.                 list.add(result1);
58.                 list.add(result2);
59.                 return list;
60.             }
61.         });
62.         System.out.println("合并结果为" + future3.get());
63.     }
64. }

复制代码

CompletableFuture08

1. package com.shaonian.juc.completable;
3. import java.util.ArrayList;
4. import java.util.List;
5. import java.util.concurrent.CompletableFuture;
6. import java.util.concurrent.ExecutionException;
7. import java.util.stream.Collectors;
9. /**
10. * 多个独立任务的合并 allOf
11. * @author 长名06
12. * @version 1.0
13. */
14. public class CompletableFuture08 {
16.     public static Integer num = 10;
18.     public static void main(String[] args) throws ExecutionException, InterruptedException {
19.         List<CompletableFuture<Integer>> list = new ArrayList<>();
21.         CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
22.             System.out.println("加10任务开启");
23.             num += 10;
24.             return num;
25.         });
26.         list.add(job1);
28.         CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
29.             System.out.println("乘10任务开启");
30.             num *= 10;
31.             return num;
32.         });
33.         list.add(job2);
35.         CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> {
36.             System.out.println("减10任务开启");
37.             num -= 10;
38.             return num;
39.         });
40.         list.add(job3);
42.         CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> {
43.             System.out.println("除10任务开启");
44.             num /= 10;
45.             return num;
46.         });
47.         list.add(job4);
48.         //使用allOf需注意，输入也会执行任务，但是无法获取到结果
49.         //allOf需要等所有的任务执行完毕
50.         /**
51.          * 返回值是CompletableFuture<Void>类型
52.          *  public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
53.          *         return andTree(cfs, 0, cfs.length - 1);
54.          *  }
55.          */
56. //        CompletableFuture<Void> allJob = CompletableFuture.allOf(list.toArray(new CompletableFuture[0]));
57. //        System.out.println(allJob.get());
58.         //也可以使用 join的形式，执行，可以获取结果
59.         List<Integer> allResult = list.stream().map(CompletableFuture::join)
60.                 .collect(Collectors.toList());
61.         System.out.println(allResult);
62.     }
63. }

复制代码

CompletableFuture09

1. package com.shaonian.juc.completable;
3. import java.util.ArrayList;
4. import java.util.List;
5. import java.util.concurrent.CompletableFuture;
6. import java.util.concurrent.ExecutionException;
8. /**
9. * anyOf
10. * @author 长名06
11. * @version 1.0
12. */
13. public class CompletableFuture09 {
14.     public static Integer num = 10;
16.     public static void main(String[] args) throws ExecutionException, InterruptedException {
17.         List<CompletableFuture<Integer>> list = new ArrayList<>();
19.         CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
20.             try {
21.                 Thread.sleep(5000);
22.             } catch (InterruptedException e) {
23.                 e.printStackTrace();
24.             }
25.             System.out.println("加10任务开启");
26.             num += 10;
27.             return num;
28.         });
29.         list.add(job1);
31.         CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
32.             try {
33.                 Thread.sleep(4000);
34.             } catch (InterruptedException e) {
35.                 e.printStackTrace();
36.             }
37.             System.out.println("乘10任务开启");
38.             num *= 10;
39.             return num;
40.         });
41.         list.add(job2);
43.         CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> {
44.             try {
45.                 Thread.sleep(3000);
46.             } catch (InterruptedException e) {
47.                 e.printStackTrace();
48.             }
49.             System.out.println("减10任务开启");
50.             num -= 10;
51.             return num;
52.         });
53.         list.add(job3);
55.         CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> {
56.             try {
57.                 Thread.sleep(2000);
58.             } catch (InterruptedException e) {
59.                 e.printStackTrace();
60.             }
61.             System.out.println("除10任务开启");
62.             num /= 10;
63.             return num;
64.         });
65.         list.add(job4);
66.         //anyOf，这里只要有一个job执行完毕，就结束所有的任务执行，不需要等待所有的job执行完毕
67.         //但是这个很鸡肋，因为如果不要执行所有的任务，就没必要开启一个CompletableFuture了
68.         //也可以适用于竞争的场景，先执行成功的获取结果，其他的不再竞争了
69.         CompletableFuture<Object> allJob = CompletableFuture.anyOf(list.toArray(new CompletableFuture[0]));
70.         System.out.println(allJob.get());
71.     }
72. }

复制代码

只是为了记录自己的学习历程，且本人水平有限，不对之处，请指正。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！