CompleteFuture 并行异步返回处理

妖狐艹你老母 2023-09-28 15:54 25阅读 0赞

### CompleteFuture 并行异步返回处理 ###

*  CompleteFuture实现
 *   *  说明：
     *  实现：
     *  结果

# CompleteFuture实现 #

## 说明： ##

对于集合数据的处理，如果每个元素的处理存在一定耗时，可以将串行（遍历）处理，改为并行处理

如：

**1.接口调用，返回值汇总**

**2.List<List< Integer >>，取出最大值**

**3.查寻数据库，返回值汇总**

## 实现： ##

package com.demo.utils;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.Collectors;
    
    /**
     * Description: 并行测试类
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-07-19 22:44
     * @version: V1.0.0
     */
    public class ParallelUtil {
        
    
    
        private static Executor moonExecutors;
    
        public static void main(String[] args) {
        
    
    
            ParallelUtil util = new ParallelUtil();
    
            moonExecutors = util.getAsyncThread();
    
    
            System.out.println(util.query());
    
        }
    
        /**
         * 查询汇总
         * @return
         */
        public int query(){
        
            Map<Integer,String> map = new HashMap<>(2);
            map.put(1,"url1");
            map.put(2,"url2");
    
            //定义任务集合
            List<CompletableFuture<Integer>> completableFutureList = new ArrayList<>(map.size());
            //初始化任务
            for (Integer key:map.keySet()){
        
                switch (key){
        
                    case 1:
                        completableFutureList.add(CompletableFuture.supplyAsync(new ParallelTaskImplV1(map.get(key))::query,moonExecutors));
                        break;
                    case 2:
                        completableFutureList.add(CompletableFuture.supplyAsync(new ParallelTaskImplV2(map.get(key))::query,moonExecutors));
                        break;
                    default:
                        break;
                }
            }
            //多任务同时返回
            CompletableFuture<Void> future = CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[map.size()]));
            //获取结果
            CompletableFuture<List<Integer>> listCompletableFuture = future.thenApplyAsync(v->{
        
                return completableFutureList.stream().map(CompletableFuture::join).collect(Collectors.toList());
            });
            //解析
            int result = 0;
            for (Integer i:listCompletableFuture.join()){
        
                //TODO 处理请求返回值
                result += i;
            }
    
            return result;
        }
    
        /**
         * 自定义线程池
         * @return
         */
        public Executor getAsyncThread() {
        
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            //线程池维护线程的最少数量，核心线程数
            taskExecutor.setCorePoolSize(5);
            //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
            taskExecutor.setMaxPoolSize(10);
            //缓存队列
            taskExecutor.setQueueCapacity(20);
            //允许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
            taskExecutor.setKeepAliveSeconds(200);
            //线程名称前缀
            taskExecutor.setThreadNamePrefix("zhx-moon-thread-");
            // 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者
            taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            // 调度器shutdown被调用时等待当前被调度的任务完成
            taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
            // 等待时长
            taskExecutor.setAwaitTerminationSeconds(60);
            taskExecutor.initialize();
            return taskExecutor;
        }
    
        /**
         * 并行任务接口类
         */
        interface ParallelTask{
        
            /**
             * 定义一个并行任务接口
             * @return
             */
            int query();
        }
    
        /**
         * 并行任务实现 1
         */
        static class ParallelTaskImplV1 implements ParallelTask{
        
    
            private String url;
    
            /**
             * 通过构造方法初始化参数
             * @param url
             */
            public ParallelTaskImplV1(String url){
        
                this.url = url;
            }
    
            @Override
            public int query() {
        
                //TODO 请求接口 1
                try {
        
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                System.out.println("Request:" + url);
                return 1;
            }
        }
    
        /**
         * 并行任务实现 2
         */
        static class ParallelTaskImplV2 implements ParallelTask{
        
    
            private String url;
    
            /**
             * 通过构造方法初始化参数
             * @param url
             */
            public ParallelTaskImplV2(String url){
        
                this.url = url;
            }
    
            @Override
            public int query() {
        
                //TODO 请求接口 2
                try {
        
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                System.out.println("Request:" + url);
                return 2;
            }
        }
    }

## 结果 ##

Connected to the target VM, address: '127.0.0.1:53099', transport: 'socket'
    Java HotSpot(TM) 64-Bit Server VM warning: Sharing is only supported for boot loader classes because bootstrap classpath has been appended
    23:52:47.448 [main] DEBUG org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService
    Request:url2
    Request:url1
    3