Java的CompletableFuture，Java的多线程开发

三、Java8的CompletableFuture，Java的多线程开发

1、CompletableFuture的常用方法

• 以后用到再加
  

1. runAsync() :开启异步（创建线程执行任务），无返回值
2. supplyAsync() :开启异步（创建线程执行任务），有返回值
3. thenApply() :然后应用，适用于有返回值的结果，拿着返回值再去处理。
4. exceptionally():用于处理异步任务执行过程中出现异常的情况的一个方法:返回默认值或者一个替代的 CompletableFuture 对象，从而避免系统的崩溃或异常处理的问题。
5. handle():类似exceptionally()
8. get()  :阻塞线程：主要可以： ①获取线程中的异常然后处理异常、②设置等待时间
9. join() :阻塞线程：推荐使用  join()  方法，因为它没有受到 interrupt 的干扰，不需要捕获异常，也不需要强制类型转换。他自己会抛出异常。
12. CompletableFuture.allOf()
13. CompletableFuture.anyOf()

复制代码

• get() 和 join() 方法区别？
  
  
  
  • 都可以阻塞线程 —— 等所有任务都执行完了再执行后续代码。
    
  
  

1. CompletableFuture 中的  get()  和  join()  方法都用于获取异步任务的执行结果，但是在使用时需要注意以下几点区别：
3. 1. 抛出异常的方式不同：如果异步任务执行过程中出现异常， get()  方法会抛出 ExecutionException 异常，而  join()  方法会抛出 CompletionException 异常，这两个异常都是继承自 RuntimeException 的。
5. 2. 方法调用限制不同： join()  方法是不可以被中断的，一旦调用就必须等待任务执行完成才能返回结果；而  get()  方法可以在调用时设置等待的超时时间，如果超时还没有获取到结果，就会抛出 TimeoutException 异常。
7. 3. 返回结果类型不同： get()  方法返回的是异步任务的执行结果，该结果是泛型类型 T 的，需要强制转换才能获取真正的结果；而  join()  方法返回的是异步任务的执行结果，该结果是泛型类型 T，不需要强制转换。
9. 4. 推荐使用方式不同：推荐在 CompletableFuture 中使用  join()  方法，因为它没有受到 interrupt 的干扰，不需要捕获异常，也不需要强制类型转换。
11. 综上所述， get()  方法和  join()  方法都是获取异步任务的执行结果，但是在使用时需要根据具体场景选择使用哪个方法。如果需要获取执行结果并且不希望被中断，推荐使用  join()  方法；如果需要控制等待时间或者需要捕获异常，则可以使用  get()  方法。

复制代码

• anyOf() 和 allOf()  的区别？
  

1. CompletableFuture 是 Java 8 引入的一个强大的异步编程工具，它支持链式调用、组合和转换异步操作等功能。其中，anyOf 和 allOf 都是 CompletableFuture 的两个常用方法，它们的区别如下：
3. 1. anyOf：任意一个 CompletableFuture 完成，它就会跟随这个 CompletableFuture 的结果完成，返回第一个完成的 CompletableFuture 的结果。
5. 2. allOf：所有的 CompletableFuture 都完成时，它才会跟随它们的结果完成，返回一个空的 CompletableFuture。
7. 简而言之，anyOf 和 allOf 的最大区别是：anyOf 任意一个 CompletableFuture 完成就跟着它的结果完成，而 allOf 所有的 CompletableFuture 完成才可以完成，并返回一个空的 CompletableFuture。
9. 举例来说，如果有三个 CompletableFuture：f1、f2、f3，其中 f1 和 f2 可能会返回一个字符串，而 f3 可能会返回一个整数，那么：
11. - anyOf(f1, f2, f3) 的结果是 f1、f2、f3 中任意一个 CompletableFuture 的结果；
12. - allOf(f1, f2, f3) 的结果是一个空的 CompletableFuture，它的完成状态表示 f1、f2、f3 是否全部完成。
14. 总之，anyOf 和 allOf 在实际使用中可以根据不同的需求来选择，它们都是 CompletableFuture 中非常强大的组合操作。

复制代码

2、使用CompletableFuture

2.1、实体类准备

1. package com.cc.md.entity;
3. import lombok.Data;
5. /**
6. * @author CC
7. * @since 2023/5/24 0024
8. */
9. @Data
10. public class UserCs {
12.     private String name;
14.     private Integer age;
16. }

复制代码

2.2、常用方式

• 无返回值推荐：开启多线程——无返回值的——阻塞：test06
  

1.     @Resource(name = "myIoThreadPool")
2.     private ThreadPoolTaskExecutor myIoThreadPool;
3.    
4.     //CompletableFuture开启多线程——无返回值的
5.     @Test
6.     public void test06() throws Exception {
7.         List<CompletableFuture<Void>> futures = new ArrayList<>();
8.         //循环，模仿很多任务
9.         for (int i = 0; i < 1000; i++) {
10.             int finalI = i;
11.             CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
12.                 //第一批创建的线程数
13.                 log.info("打印：{}", finalI);
14.                 //模仿io流耗时
15.                 try {
16.                     Thread.sleep(5000);
17.                 } catch (InterruptedException e) {
18.                     throw new RuntimeException(e);
19.                 }
20.             }, myIoThreadPool);
21.             futures.add(future);
22.         }
23.         //阻塞：多线程的任务执行。相当于多线程执行完了，再执行后面的代码
24.         //如果不阻塞，上面的相当于异步执行了。
25.         //阻塞方式1：可以获取返回的异常、设置等待时间
26. //        futures.forEach(future -> {
27. //            try {
28. //                future.get();
29. //            } catch (Exception e) {
30. //                throw new RuntimeException(e);
31. //            }
32. //        });
33.         //阻塞方式2（推荐）
34.         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
35.         log.info("打印：都执行完了。。。");
36.     }

复制代码

• 有返回值推荐：开启多线程——有返回值的，返回一个新的List——阻塞——使用stream流的map：test09
  
  
  
  • test07、test08 可以转化为 test09 (现在这个)
    
  • 可以返回任务类型的值，不一定要返回下面的user对象。
    
  
  

1.     @Resource(name = "myIoThreadPool")
2.     private ThreadPoolTaskExecutor myIoThreadPool;
3.    
4.     //CompletableFuture开启多线程——有返回值的，返回一个新的List——先有数据的情况——使用stream流的map
5.     //像这种，需要构建另一个数组的，相当于一个线程执行完了，会有返回值
6.     //使用stream流的map + CompletableFuture.supplyAsync()
7.     @Test
8.     public void test09() throws Exception {
9.         //先获取数据，需要处理的任务。
10.         List<UserCs> users = this.getUserCs();
11.         //莫法处理任务
12.         List<CompletableFuture<UserCs>> futures = users.stream()
13.                 .map(user -> CompletableFuture.supplyAsync(() -> {
14.                     // 处理数据
15.                     user.setName(user.getName() + "-改");
16.                     log.info("打印-改：{}", user.getName());
17.                     // 其他的业务逻辑。。。
19.                     return user;
20.                 }, myIoThreadPool)).collect(Collectors.toList());
22.         //获取futures
23.         List<UserCs> endList = futures.stream()
24.                 //阻塞所有线程
25.                 .map(CompletableFuture::join)
26.                 //取age大于10的用户
27.                 .filter(user -> user.getAge() > 10)
28.                 //按照age升序排序
29.                 .sorted(Comparator.comparing(UserCs::getAge))
30.                 .collect(Collectors.toList());
31.         log.info("打印：都执行完了。。。{}", endList);
32.     }

复制代码

2.3、异常处理

• exceptionally
  
• handle
  

1.         //CompletableFuture 异常处理
2.     @Test
3.     public void test10() throws Exception {
4.         //先获取数据，需要处理的任务。
5.         List<UserCs> users = this.getUserCs();
6.         //莫法处理任务
7.         List<CompletableFuture<UserCs>> futures = users.stream()
8.                 .map(user -> CompletableFuture.supplyAsync(() -> {
9.                         if (user.getAge() > 5){
10.                             int a = 1/0;
11.                         }
12.                         // 处理数据
13.                         user.setName(user.getName() + "-改");
14.                         log.info("打印-改：{}", user.getName());
15.                         // 其他的业务逻辑。。。
17.                         return user;
18.                     }, myIoThreadPool)
19.                     //处理异常方式1：返回默认值或者一个替代的 Future 对象，从而避免系统的崩溃或异常处理的问题。
20.                     .exceptionally(throwable -> {
21.                         //可以直接获取user
22.                         System.out.println("异常了：" + user);
23.                         //处理异常的方法……
24.                         //1还可以进行业务处理……比如将异常数据存起来，然后导出……
25.                         //2返回默认值，如：user、null
26.                         //return user;
27.                         //3抛出异常
28.                         throw new RuntimeException(throwable.getMessage());
29.                     })
30.                     //处理异常方式2：类似exceptionally（不推荐）
31. //                    .handle((userCs, throwable) -> {
32. //                        System.out.println("handle：" + user);
33. //                        if (throwable != null) {
34. //                            // 处理异常
35. //                            log.error("处理用户信息出现异常，用户名为：" + user.getName(), throwable);
36. //                            // 返回原始数据
37. //                            return userCs;
38. //                        } else {
39. //                            // 返回正常数据
40. //                            return userCs;
41. //                        }
42. //                    })
43.                 )
44.                 .collect(Collectors.toList());
46.         //获取futures
47.         List<UserCs> endList = futures.stream()
48.                 //阻塞所有线程
49.                 .map(CompletableFuture::join)
50.                 //取age大于10的用户
51.                 .filter(user -> user.getAge() > 10)
52.                 //按照age升序排序
53.                 .sorted(Comparator.comparing(UserCs::getAge))
54.                 .collect(Collectors.toList());
55.         log.info("打印：都执行完了。。。{}", endList);
56.     }

复制代码

2.4、CompletableFuture的使用测试

1、推荐使用：test03、test05、test09、test10、test11
2、test07、test08就是test09的前身。

• test01：获取当前电脑（服务器）的cpu核数
  
• test02：线程池原始的使用（不推荐直接这样用）
  
• test03：开启异步1 —— @Async
  
• test04：开启异步2 —— CompletableFuture.runAsync()
  
• test05：开启异步2的改造 —— CompletableFuture.runAsync() 和 supplyAsync()  —— 阻塞所有异步方法，一起提交
  
  
  
  • 1. 相当于开了3个线程去执行三个不同的方法，然后执行完后一起提交。

    复制代码
  
  
• test052：开启异步2的改造 —— 第一个任务执行完了，获取到返回值，给后面的执行，可以连写，也可以单写。 —— 阻塞线程：get、join
  
• test06：CompletableFuture开启多线程——无返回值的
  
• test07：CompletableFuture开启多线程——无返回值的——构建一个新List
  
  
  
  • 1. 1、相当于多线程执行任务，然后把结果插入到List中
    2. 2、接收多线程的List必须是线程安全的，ArrayList线程不安全
    3.    线程安全的List —— CopyOnWriteArrayList 替代 ArrayList

    复制代码
  
  
• test08：CompletableFuture开启多线程——无返回值的——构建一个新List——先有数据的情况（基本和test07是一个方法）
  
• test09：CompletableFuture开启多线程——有返回值的，返回一个新的List——先有数据的情况——使用stream流的map
  
• test10：CompletableFuture 异常处理。相当于是 test09的增强，处理异常
  
• test11：CompletableFuture 异常处理：如果出现异常就舍弃任务。
  
  
  
  • 1. 1、想了一下，出现异常后的任务确实没有执行下去了，任务不往下执行，怎么会发现异常呢？
    2. 2、发现了异常任务也就完了。而且打印了异常，相当于返回了异常。
    3. 3、未发生异常的任务会执行完成。如果发生异常都返回空，最后舍弃空的，就得到任务执行成功的 CompletableFuture

    复制代码
  
  

↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓所有方式↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓

1. package com.cc.md;
3. import com.cc.md.entity.UserCs;
4. import com.cc.md.service.IAsyncService;
5. import org.junit.jupiter.api.Test;
6. import org.slf4j.Logger;
7. import org.slf4j.LoggerFactory;
8. import org.springframework.boot.test.context.SpringBootTest;
9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
11. import javax.annotation.Resource;
12. import java.util.*;
13. import java.util.concurrent.CompletableFuture;
14. import java.util.concurrent.CopyOnWriteArrayList;
15. import java.util.concurrent.ForkJoinPool;
16. import java.util.concurrent.TimeUnit;
17. import java.util.stream.Collectors;
19. @SpringBootTest
20. class Test01 {
22.     private static final Logger log = LoggerFactory.getLogger(Test01.class);
24.     @Resource(name = "myIoThreadPool")
25.     private ThreadPoolTaskExecutor myIoThreadPool;
26.     /**
27.      * 异步类
28.      */
29.     @Resource
30.     private IAsyncService asyncService;
32.     @Test
33.     void test01() {
34.         //获取当前jdk能调用的CPU个数（当前服务器的处理器个数）
35.         int i = Runtime.getRuntime().availableProcessors();
36.         System.out.println(i);
37.     }
39.     //线程池原始的使用
40.     @Test
41.     void test02() {
42.         try {
43.             for (int i = 0; i < 1000; i++) {
44.                 int finalI = i;
45.                 myIoThreadPool.submit(() -> {
46.                     //第一批创建的线程数
47.                     log.info("打印：{}", finalI);
48.                     //模仿io流耗时
49.                     try {
50.                         Thread.sleep(5000);
51.                     } catch (InterruptedException e) {
52.                         throw new RuntimeException(e);
53.                     }
54.                 });
55.             }
56.         }catch(Exception e){
57.             throw new RuntimeException(e);
58.         }finally {
59.             myIoThreadPool.shutdown();
60.         }
61.     }
63.     //开启异步1 —— @Async
64.     @Test
65.     public void test03() throws Exception {
66.         log.info("打印：{}", "异步测试的-主方法1");
67.         asyncService.async1();
68.         asyncService.async2();
69.         //不会等待异步方法执行，直接返回前端数据
70.         log.info("打印：{}", "异步测试的-主方法2");
71.     }
73.     //开启异步2 —— CompletableFuture.runAsync()
74.     @Test
75.     public void test04() throws Exception {
76.         log.info("打印：{}", "异步测试的-主方法1");
77.         CompletableFuture.runAsync(() -> {
78.             log.info("打印：{}", "异步方法1！");
79.             //异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
80.             this.async2("异步方法1！-end");
81.         }, myIoThreadPool);
82.         //不会等待异步方法执行，直接返回前端数据
83.         log.info("打印：{}", "异步测试的-主方法2");
84.     }
86.     //异步需要执行的方法，可以写在同一个类中。
87.     private void async2(String msg) {
88.         //模仿io流耗时
89.         try {
90.             Thread.sleep(5000);
91.         } catch (InterruptedException e) {
92.             throw new RuntimeException(e);
93.         }
94.         log.info("打印：{}", msg);
95.     }
97.     //开启异步2的改造 —— CompletableFuture.runAsync() 和 supplyAsync()  —— 阻塞所有异步方法，一起提交
98.     //相当于开了3个线程去执行三个不同的方法，然后执行完后一起提交。
99.     @Test
100.     public void test05() throws Exception {
101.         log.info("打印：{}", "异步测试的-主方法1");
102.         //异步执行1
103.         CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
104.             log.info("打印：{}", "异步方法1！");
105.             //异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
106.             this.async2("异步方法1-end");
107.             return "异步方法1";
108.         }, myIoThreadPool);
110.         //异步执行2
111.         CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
112.             log.info("打印：{}", "异步方法2！");
113.             //异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
114.             this.async2("异步方法2-end");
115.             return "异步方法2";
116.         }, myIoThreadPool);
118.         //异步执行3，不用我们自己的线程池 —— 用的就是系统自带的 ForkJoinPool 线程池
119.         CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
120.             log.info("打印：{}", "异步方法3！");
121.             //异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
122.             this.async2("异步方法3-end");
123.         });
125.         //阻塞所有异步方法，一起提交后才走下面的代码
126.         CompletableFuture.allOf(future1, future2, future3).join();
128.         log.info("打印：{}", "异步-阻塞-测试的-主方法2-end");
129.     }
131.     //开启异步2的改造 —— 第一个任务执行完了，获取到返回值，给后面的执行，可以连写，也可以单写。 —— 阻塞线程：get、join
132.     // CompletableFuture 的 get 和 join 方法区别：
133.     // get：①可以获取线程中的异常、②设置等待时间
134.     // join：推荐在 CompletableFuture 中使用  join()  方法，因为它没有受到 interrupt 的干扰，不需要捕获异常，也不需要强制类型转换。
135.     @Test
136.     public void test052() throws Exception {
137.         log.info("打印：{}", "异步测试的-主方法1");
138.         //异步执行1
139.         CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
140.             log.info("打印：{}", "异步方法1！");
141.             // 异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
142.             String str = "异步方法1-end";
143.             this.async2(str);
144.             return str;
145.         }, myIoThreadPool);
147.         // 异步执行2 - 无返回值 —— 分开写的方式
148.         CompletableFuture<Void> future2 = future1.thenAccept(str1 -> {
149.             log.info("打印：{}", "异步方法2！");
150.             // 异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
151.             this.async2(String.format("%s-加-异步方法2！ - 无返回值 - ",str1));
152.         });
154.         // 异步执行3 - 有返回值 —— 分开写future1，连写future3方式
155.         CompletableFuture<String> future3 = future1.thenApply(str2 -> {
156.             log.info("打印：{}", "异步方法3！");
157.             // 异步执行的代码，也可以是方法，该方法不用单独写到其他类中。
158.             this.async2(String.format("%s-加-异步方法3！ - 有返回值 - ", str2));
159.             return "异步执行3 - 有返回值 ";
161.             //连写的方式。
162.         }).thenApply(str3 -> {
163.             String format = String.format("%s- end", str3);
164.             log.error("异步3然后应用 - {}", format);
165.             //返回后面的应用
166.             return format;
167.         });
168.         // 获取future3的返回值：
169.         //如果需要捕获异常、设置等待超时时间，则用get
170.         log.info("future3的返回值(不阻塞)：{}", future3.get());
171. //        log.info("future3的返回值(不阻塞-设置等待时间，超时报错：TimeoutException)：{}",
172. //                future3.get(2, TimeUnit.SECONDS));
173.         //推荐使用 join方法
174. //        log.info("future3的返回值(阻塞)：{}", future3.join());
176.         //阻塞所有异步方法，一起提交后才走下面的代码
177.         CompletableFuture.allOf(future1, future2).join();
179.         log.info("打印：{}", "异步-阻塞-测试的-主方法2-end");
180.     }
182.     //CompletableFuture开启多线程——无返回值的
183.     @Test
184.     public void test06() throws Exception {
185.         List<CompletableFuture<Void>> futures = new ArrayList<>();
186.         //循环，模仿很多任务
187.         for (int i = 0; i < 1000; i++) {
188.             int finalI = i;
189.             CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
190.                 //第一批创建的线程数
191.                 log.info("打印：{}", finalI);
192.                 //模仿io流耗时
193.                 try {
194.                     Thread.sleep(5000);
195.                 } catch (InterruptedException e) {
196.                     throw new RuntimeException(e);
197.                 }
198.             }, myIoThreadPool);
199.             futures.add(future);
200.         }
201.         //阻塞：多线程的任务执行。相当于多线程执行完了，再执行后面的代码
202.         //如果不阻塞，上面的相当于异步执行了。
203.         //阻塞方式1：可以获取返回的异常、设置等待时间
204. //        futures.forEach(future -> {
205. //            try {
206. //                future.get();
207. //            } catch (Exception e) {
208. //                throw new RuntimeException(e);
209. //            }
210. //        });
211.         //阻塞方式2（推荐）
212.         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
213.         log.info("打印：都执行完了。。。");
214.     }
216.     //CompletableFuture开启多线程——无返回值的——构建一个新List
217.     //相当于多线程执行任务，然后把结果插入到List中
218.     //接收多线程的List必须是线程安全的，ArrayList线程不安全
219.     //线程安全的List —— CopyOnWriteArrayList 替代 ArrayList
220.     @Test
221.     public void test07() throws Exception {
222.         List<CompletableFuture<Void>> futures = new ArrayList<>();
223.         //存数据的List
224.         List<UserCs> addList = new CopyOnWriteArrayList<>();
225.         //循环，模仿很多任务
226.         for (int i = 0; i < 1000; i++) {
227.             int finalI = i;
228.             CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
229.                 log.info("打印：{}", finalI);
230.                 UserCs userCs = new UserCs();
231.                 userCs.setName(String.format("姓名-%s", finalI));
232.                 userCs.setAge(finalI);
233.                 addList.add(userCs);
234.             }, myIoThreadPool);
235.             futures.add(future);
236.         }
237.         //阻塞
238.         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
240.         //返回新的List：endList，取age大于10的用户
241.         List<UserCs> endList = addList.stream()
242.                 .filter(user -> user.getAge() > 10)
243.                 //按照age升序排序
244.                 .sorted(Comparator.comparing(UserCs::getAge))
245.                 .collect(Collectors.toList());
246.         log.info("打印：都执行完了。。。{}", endList);
247.     }
249.     //CompletableFuture开启多线程——无返回值的——构建一个新List——先有数据的情况
250.     //用CopyOnWriteArrayList 替代 ArrayList接收
251.     @Test
252.     public void test08() throws Exception {
253.         //先获取数据，需要处理的任务。
254.         List<UserCs> users = this.getUserCs();
255.         //开启多线程
256.         List<CompletableFuture<Void>> futures = new ArrayList<>();
257.         //存数据的List
258.         List<UserCs> addList = new CopyOnWriteArrayList<>();
259.         //莫法处理任务
260.         users.forEach(user -> {
261.             CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
262.                 //添加数据
263.                 user.setName(user.getName() + "-改");
264.                 addList.add(user);
266.                 log.info("打印-改：{}", user.getName());
267.                 //其他的业务逻辑。。。
269.             }, myIoThreadPool);
270.             futures.add(future);
271.         });
273.         //阻塞
274.         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
276.         //返回新的List：endList
277.         List<UserCs> endList = addList.stream()
278.                 .filter(user -> user.getAge() > 10)
279.                 //按照age升序排序
280.                 .sorted(Comparator.comparing(UserCs::getAge))
281.                 .collect(Collectors.toList());
282.         log.info("打印：都执行完了。。。{}", endList);
283.     }
285.     //CompletableFuture开启多线程——有返回值的，返回一个新的List——先有数据的情况——使用stream流的map
286.     //像这种，需要构建另一个数组的，相当于一个线程执行完了，会有返回值
287.     //使用stream流的map + CompletableFuture.supplyAsync()
288.     @Test
289.     public void test09() throws Exception {
290.         //先获取数据，需要处理的任务。
291.         List<UserCs> users = this.getUserCs();
292.         //莫法处理任务
293.         List<CompletableFuture<UserCs>> futures = users.stream()
294.                 .map(user -> CompletableFuture.supplyAsync(() -> {
295.                     // 处理数据
296.                     user.setName(user.getName() + "-改");
297.                     log.info("打印-改：{}", user.getName());
298.                     // 其他的业务逻辑。。。
300.                     return user;
301.                 }, myIoThreadPool)).collect(Collectors.toList());
303.         //获取futures
304.         List<UserCs> endList = futures.stream()
305.                 //阻塞所有线程
306.                 .map(CompletableFuture::join)
307.                 //取age大于10的用户
308.                 .filter(user -> user.getAge() > 10)
309.                 //按照age升序排序
310.                 .sorted(Comparator.comparing(UserCs::getAge))
311.                 .collect(Collectors.toList());
312.         log.info("打印：都执行完了。。。{}", endList);
313.     }
315.     //基础数据
316.     private List<UserCs> getUserCs() {
317.         List<UserCs> users = new ArrayList<>();
318.         for (int i = 0; i < 1000; i++) {
319.             UserCs userCs = new UserCs();
320.             userCs.setName(String.format("姓名-%s", i));
321.             userCs.setAge(i);
322.             users.add(userCs);
323.         }
324.         return users;
325.     }
327.     //CompletableFuture 异常处理
328.     @Test
329.     public void test10() throws Exception {
330.         //先获取数据，需要处理的任务。
331.         List<UserCs> users = this.getUserCs();
332.         //莫法处理任务
333.         List<CompletableFuture<UserCs>> futures = users.stream()
334.                 .map(user -> CompletableFuture.supplyAsync(() -> {
335.                         if (user.getAge() > 5){
336.                             int a = 1/0;
337.                         }
338.                         // 处理数据
339.                         user.setName(user.getName() + "-改");
340.                         log.info("打印-改：{}", user.getName());
341.                         // 其他的业务逻辑。。。
343.                         return user;
344.                     }, myIoThreadPool)
345.                     //处理异常方式1：返回默认值或者一个替代的 Future 对象，从而避免系统的崩溃或异常处理的问题。
346.                     .exceptionally(throwable -> {
347.                         //可以直接获取user
348.                         System.out.println("异常了：" + user);
349.                         //处理异常的方法……
350.                         //1还可以进行业务处理……比如将异常数据存起来，然后导出……
351.                         //2返回默认值，如：user、null
352.                         //return user;
353.                         //3抛出异常
354.                         throw new RuntimeException(throwable.getMessage());
355.                     })
356.                     //处理异常方式2：类似exceptionally（不推荐）
357. //                    .handle((userCs, throwable) -> {
358. //                        System.out.println("handle：" + user);
359. //                        if (throwable != null) {
360. //                            // 处理异常
361. //                            log.error("处理用户信息出现异常，用户名为：" + user.getName(), throwable);
362. //                            // 返回原始数据
363. //                            return userCs;
364. //                        } else {
365. //                            // 返回正常数据
366. //                            return userCs;
367. //                        }
368. //                    })
369.                 )
370.                 .collect(Collectors.toList());
372.         //获取futures
373.         List<UserCs> endList = futures.stream()
374.                 //阻塞所有线程
375.                 .map(CompletableFuture::join)
376.                 //取age大于10的用户
377.                 .filter(user -> user.getAge() > 10)
378.                 //按照age升序排序
379.                 .sorted(Comparator.comparing(UserCs::getAge))
380.                 .collect(Collectors.toList());
381.         log.info("打印：都执行完了。。。{}", endList);
382.     }
384.     //CompletableFuture 异常处理：如果出现异常就舍弃任务。
385.     // 想了一下，出现异常后的任务确实没有执行下去了，任务不往下执行，怎么会发现异常呢？
386.     // 发现了异常任务也就完了。而且打印了异常，相当于返回了异常。
387.     // 未发生异常的任务会执行完成。如果发生异常都返回空，最后舍弃空的，就得到任务执行成功的 CompletableFuture
388.     @Test
389.     public void test11() {
390.         List<UserCs> users = getUserCs();
391.         List<CompletableFuture<UserCs>> futures = users.stream()
392.                 .map(user -> CompletableFuture.supplyAsync(() -> {
393.                             if (user.getAge() > 15) {
394.                                 int a = 1 / 0;
395.                             }
396.                             user.setName(user.getName() + "-改");
397.                             log.info("打印-改：{}", user.getName());
398.                             return user;
399.                         }, myIoThreadPool)
400.                         //处理异常
401.                         .exceptionally(throwable -> {
402.                             //其他处理异常的逻辑
404.                             return null;
405.                         })
406.                 )
407.                 //舍弃返回的对象是null的 CompletableFuture
408.                 .filter(e -> Objects.nonNull(e.join())).collect(Collectors.toList());
410.         //获取futures
411.         List<UserCs> endList = futures.stream()
412.                 //阻塞所有线程
413.                 .map(CompletableFuture::join)
414.                 //取age大于10的用户
415.                 .filter(user -> user.getAge() > 10)
416.                 //按照age升序排序
417.                 .sorted(Comparator.comparing(UserCs::getAge))
418.                 .collect(Collectors.toList());
419.         log.info("打印：都执行完了。。。{}", endList);
421.     }
423. }

复制代码

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