CompletableFuture实现线程池结果

忘是亡心i 2024-04-03 10:55 78阅读 0赞

### 背景 ###

向线程池中提交任务的submit方法不是阻塞方法，而Future.get方法是一个阻塞方法，当submit提交多个任务时，只有所有任务都完成后，才能使用get按照任务的提交顺序得到返回结果。  
Future虽然可以实现获取异步执行结果的需求，但是它没有提供通知的机制，要么使用阻塞，在future.get()的地方等待future返回的结果，这时又变成同步操作；要么使用isDone()轮询地判断Future是否完成，这样会耗费CPU的资源

### 原本代码 ###

1.首先使用spring自带的线程池ThreadPoolTaskExecutor ，我们对他先定义一些属性覆盖自带属性

@Configuration
    @EnableAsync
    public class ExecutorConfig {
        
        @Bean("executor")
        public Executor executor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            // 设置核心线程数
            executor.setCorePoolSize(3);
            // 设置最大线程数
            executor.setMaxPoolSize(20);
            //配置队列大小
            executor.setQueueCapacity(6);
            // 设置线程活跃时间（秒）
            executor.setKeepAliveSeconds(60);
            // 设置默认线程名称
            executor.setThreadNamePrefix("线程池_");
            // 等待所有任务结束后再关闭线程池
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //执行初始化
            executor.initialize();
            return executor;
        }
    }

2.在写个控制层用于访问服务层方法

@RestController
    @Slf4j
    public class Hello {
    
        @Autowired
        private AsyncService asyncService;
    
        @RequestMapping("/")
        public String submit(){
    
            //调用service层的任务
            asyncService.executeAsync();
    
            return "success";
        }
    }

3 服务层接口

public interface AsyncService {
    
        /**
         * 执行异步任务
         */
        void executeAsync();
    }

**实现类**，主要是循环使用线程池执行Callable实现类（下面），传入i作为标记，查看打印结果是否是按顺序打印，加了@Async和@Scheduled的类都会，都会使用线程池里的线程异步执行

@Service
    @Slf4j
    public class AsyncServiceImpl implements AsyncService {
    
        @Resource(name = "executor")
        private ThreadPoolTaskExecutor executor;
    
        private List<Future<String>> futures = new ArrayList<Future<String>>();
    
        @Override
        @Async
        public void executeAsync() {
            log.info("开始线程池");
            try {
                for (int i = 0; i < 10; i++) {
                    futures.add(executor.submit(new GetStrService(i)));
                }
    
                for (Future<String> future : futures) {
                    while (true) {
                        if (future.isDone()) {
                            System.out.println(future.get());
                            break;
                        } else {
                            //每次轮询休息1毫秒（CPU纳秒级），避免CPU高速轮循耗空CPU
                            Thread.sleep(1);
                        }
                    }
                }
            } catch (Exception e) {
                log.error("异常", e);
            }
            log.info("结束线程池");
        }
    }

4.**任务类**：需要执行并返回结果的线程任务，任务加了随机延时，用来测试`future.get()`方法，前面的线程最晚才返回结果，然后后面的是很快就返回，但是还是得等待前面返回了，后面才得以执行

public class GetStrService implements Callable<String> {
        private int i;
    
        public GetStrService(int i) {
            this.i = i;
        }
    
        @Override
        public String call() throws Exception {
            int t=(int) (Math.random()*(10-1)+1);
            System.out.println("第"+i+"个任务开始啦："+Thread.currentThread().getName()+"准备延时"+t+"秒");
            Thread.sleep(t*1000);
            return "第"+i+"个GetStrService任务使用的线程："+Thread.currentThread().getName();
        }
    }

5.**结果**

2021-03-18 11:45:35.486  INFO 27688 --- [          线程池_1] c.y.t.service.impl.AsyncServiceImpl      : 开始线程池
    第1个任务开始啦：线程池_3，准备延时4秒
    第0个任务开始啦：线程池_2，准备延时6秒
    第8个任务开始啦：线程池_4，准备延时2秒
    第9个任务开始啦：线程池_5，准备延时5秒
    第2个任务开始啦：线程池_4，准备延时7秒
    第3个任务开始啦：线程池_3，准备延时4秒
    第4个任务开始啦：线程池_5，准备延时9秒
    第5个任务开始啦：线程池_2，准备延时8秒
    第0个GetStrService任务使用的线程：线程池_2
    第1个GetStrService任务使用的线程：线程池_3
    第6个任务开始啦：线程池_3，准备延时3秒
    第2个GetStrService任务使用的线程：线程池_4
    第3个GetStrService任务使用的线程：线程池_3
    第7个任务开始啦：线程池_4，准备延时8秒
    第4个GetStrService任务使用的线程：线程池_5
    第5个GetStrService任务使用的线程：线程池_2
    第6个GetStrService任务使用的线程：线程池_3
    第7个GetStrService任务使用的线程：线程池_4
    第8个GetStrService任务使用的线程：线程池_4
    第9个GetStrService任务使用的线程：线程池_5
    2021-03-18 11:45:52.495  INFO 27688 --- [          线程池_1] c.y.t.service.impl.AsyncServiceImpl      : 结束线程池

我们来分析下结果，打印结果：第n个任务开始啦：线程池\_i，准备延时j秒，这个的打印结果顺序没有按照0-9来，因为for循环里提交很快，具体哪个线程会先开始看CPU的情况，所以打印语句无顺序，所以不用理他，我们只要知道执行循环submit，线程池会记录哪个线程先提交的就行。  
然后重点看，返回准备延时j秒，和返回结果语句，可以看到第0个线程是6秒后才返回，第一个线程是4秒后才返回，比第0个快，但是看返回结果却是按顺序打印，第0个GetStrService，第1个GetStrService。  
所以得出结论： future.get()方法会阻塞，获取先提交的线程的返回结果

### 改进代码 ###

使用**`CompletableFuture`**，先有返回结果的就能先获取到  
上代码  
线程池配置依旧如上

1.**任务类**  
有改动，之前实现`Callable`，现在实现java8的`Supplier`（生产者），重写`get`方法，里面内容跟上面的一样，也是用来标记查看是否按顺序获取

@Slf4j
    public class NewGetStrService implements Supplier<String> {
        private int i;
    
        public NewGetStrService(int i) {
            this.i = i;
        }
    
        @Override
        public String get() {
            try {
                int t=(int) (Math.random()*(10-1)+1);
                System.out.println("第"+i+"个任务开始啦："+Thread.currentThread().getName()+"，准备延时"+t+"秒");
                Thread.sleep(t*1000);
            } catch (Exception e) {
                log.error("GetStrNewService 异常",e);
            }
            return "第"+i+"个GetStrService任务使用的线程："+Thread.currentThread().getName();
        }
    }

2.**服务层**  
实现类：使用`CompletableFuture`实现

@Service
    @Slf4j
    public class NewAsyncServiceImpl implements AsyncService {
    
        @Resource(name = "executor")
        private ThreadPoolTaskExecutor executor;
    
        private List<Future<String>> futures = new ArrayList<Future<String>>();
    
        @Override
        @Async
        public void executeAsync() {
            log.info("开始线程池");
            for (int i = 0; i < 10; i++) {
                CompletableFuture.supplyAsync(new NewGetStrService(i), executor).whenComplete((result,e) -> {
                    //执行线程执行完以后的操作。
                    System.out.println(result);
                }).exceptionally((e) -> {
                    //抛出异常
                    System.out.println("exception " + e);
                    return "exception";
                });
            }
            log.info("结束线程池");
        }
    }

3.**控制层**  
就变了显示引用`AsyncService`的实现类，因为加上上面原先的`AsyncServiceImpl`就是俩了，得指定下

@RestController
    @Slf4j
    public class Hello {
    
        @Resource(name = "newAsyncServiceImpl")
        private AsyncService asyncService;
    
        @RequestMapping("/")
        public String submit(){
    
            //调用service层的任务
            asyncService.executeAsync();
    
            return "success";
        }
    }

4.**结果**  
可以看出执行时间任务最短的，先输出了，而不是按照提交顺序输出结果

![20210318143552270.jpg_x-oss-process_image][]

[20210318143552270.jpg_x-oss-process_image]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/03/1e238d4678b6419fb9921ef36b3fc6c5.jpg