java使用自定义线程池跑多线程,Future 不同的线程返回不同的数据,已经多线程的本地线程变量使用以及CompletableFuture使用

缺乏、安全感 2023-06-07 03:10 7阅读 0赞

使用线程池跑多线程

//创建子线程可见的本机线程变量  
InheritableThreadLocal inheritableThreadLocal = new InheritableThreadLocal<>();  
//创建子线程不可见的本机线程变量  
ThreadLocal threadLocal = new ThreadLocal<>();

InheritableThreadLocal可让子线程获取主线程数据

四种线程池创建的方式  
newCachedThreadPool  
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

newFixedThreadPool  
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

newScheduledThreadPool  
创建一个定长线程池，支持定时及周期性任务执行。

newSingleThreadExecutor  
创建一个单线程化的线程池执行任务。

/* * Copyright 2019 Wicrenet, Inc. All rights reserved. */
    package com.xy;
    
    import java.util.ArrayList;
    import java.util.concurrent.*;
    
    /** * 【】 * * @author YJX * Created on 2019/10/12 10:53 */
    
    public class ThreadPoolExecutor { 
    
        //使用 FixedThreadPool线程池
        private static final ExecutorService threadPool = Executors.newFixedThreadPool(10000, (r) -> { 
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        });
    
        //创建子线程可见的本机线程变量
        private final static InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();
        //创建子线程不可见的本机线程变量
        private final static ThreadLocal<String>            threadLocal            = new ThreadLocal<>();
    
        /** * 使用线程池获取多个线程的返回数据 */
        public static void main(String[] argv) { 
            long l = System.currentTimeMillis();
            //设置子线程可见的线程本地变量
            inheritableThreadLocal.set("yjx-inheritableThreadLocal-子线程可见");
            //设置线程本地变量
            threadLocal.set("yjx-threadLocal-子线程不可见");
            /*可用线程数量的固定线程池*/
            ExecutorService executorService = threadPool;
            ArrayList<Object> list = new ArrayList<>();
            //向线程池中添加线程
            Future thread1 = executorService.submit(new ThreadRV1());
            Future thread2 = executorService.submit(new ThreadRV2());
            Future thread3 = executorService.submit(new ThreadRV3());
            //...可以加N个线程任务
    
            executorService.shutdown();//一定要调用这个方法，不然executorService.isTerminated()永远不为true
            while (true) { //等待所有任务都结束了继续执行
                try { 
                    if (executorService.isTerminated()) { 
                        System.out.println("所有的子线程都结束了！");
                        break;
                    }
                    Thread.sleep(100);
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
    
            try { 
                list.add(thread1.get());
                list.add(thread2.get());
                list.add(thread3.get());
                System.out.println("所有线程共耗时:" + (System.currentTimeMillis() - l));
                System.out.println("list = " + list);
    
            } catch (InterruptedException | ExecutionException e) { 
                e.printStackTrace();
            }
        }
    
        public static class ThreadRV1 implements Callable { 
            @Override
            public String call() throws Exception { 
                long l = System.currentTimeMillis();
    
                String s = inheritableThreadLocal.get();
                System.out.println("t1s = " + s);
                String s1 = threadLocal.get();
                System.out.println("t1s1 = " + s1);
    
                System.out.println("线程1开始");
                TimeUnit.SECONDS.sleep(5);
                System.out.println("线程1结束-耗时" + (System.currentTimeMillis() - l));
                return "t1" + s;
            }
        }
    
    
    
        public static class ThreadRV2 implements Callable { 
            @Override
            public String call() throws Exception { 
                long l = System.currentTimeMillis();
                String s = inheritableThreadLocal.get();
                System.out.println("t1s = " + s);
                String s1 = threadLocal.get();
                System.out.println("t1s1 = " + s1);
    
                System.out.println("线程2开始");
                TimeUnit.SECONDS.sleep(5);
                System.out.println("线程2结束-耗时" + (System.currentTimeMillis() - l));
                return "t2" + s1;
            }
        }
    
        //返回Integer类型
        public static class ThreadRV3 implements Callable { 
            @Override
            public Integer call() throws Exception { 
                long l = System.currentTimeMillis();
    
                String s = inheritableThreadLocal.get();
                System.out.println("t1s = " + s);
                String s1 = threadLocal.get();
                System.out.println("t1s1 = " + s1);
    
                System.out.println("线程3开始");
                TimeUnit.SECONDS.sleep(5);
                System.out.println("线程3结束-耗时" + (System.currentTimeMillis() - l));
                return 1;
            }
        }
    }

例子2

package com.xy.pay.main.config;
    
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    
    /** * 线程池 配置 * * @author YJX * @date 2019/10/12 15:17 */
    @Configuration
    @EnableAsync
    public class ExecutorConfig { 
        //@Autowired
        //private TraceableThreadFactory traceableThreadFactory;
    
    
        /** * 默认的线程池 * * @return */
        @Bean
        @Primary
        @Qualifier("defaultExecutor")
        public Executor defaultExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(5);/*核心线程数*/
            executor.setMaxPoolSize(10);/*最大线程数*/
            executor.setQueueCapacity(10000);/*队列大小*/
            executor.setKeepAliveSeconds(60);/* 某线程空闲超过1分钟，就回收该线程*/
            //executor.setAllowCoreThreadTimeOut(true); // KeepAliveSeconds 设置也作用于【核心线程数】
            executor.setThreadNamePrefix("defaultExecutor-");
            //executor.setThreadFactory(traceableThreadFactory);
            executor.setAwaitTerminationSeconds(5);
            executor.setRejectedExecutionHandler(null);
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.initialize();
            return executor;
        }
    
        /** * 对账的线程池 * * @return */
        @Bean
        @Qualifier("executor")
        public Executor reconciliationExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(3);/*核心线程数*/
            executor.setMaxPoolSize(3);/*最大线程数*/
            executor.setQueueCapacity(10000);/*队列大小*/
            executor.setKeepAliveSeconds(60 * 3);/* 某线程空闲超过3分钟，就回收该线程*/
            executor.setAllowCoreThreadTimeOut(true);   // KeepAliveSeconds 设置也作用于【核心线程数】
            executor.setThreadNamePrefix("executor-");
            //executor.setThreadFactory(traceableThreadFactory);
            executor.setRejectedExecutionHandler(null);
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.setAwaitTerminationSeconds(5);
            executor.initialize();
            return executor;
        }
    
    }

@Autowired
        @Qualifier("executor")
        private Executor executor;
    
        /** * @return */
        @GetMapping("thread")
        public void store() { 
    
            System.out.println("例子1");
            long startTime = System.currentTimeMillis();
            ThreadPoolTaskExecutor executor1 = (ThreadPoolTaskExecutor) this.executor;
            Future<String> submit1 = executor1.submit(this::a1);
            Future<String> submit2 = executor1.submit(this::a2);
    
            ArrayList<Object> list1 = new ArrayList<>();
            try { 
                String s1 = submit1.get();
                String s2 = submit2.get();
                list1.add(s1);
                list1.add(s2);
            } catch (Exception e) { 
                throw new RuntimeException(e);
            }
            long endTime1 = System.currentTimeMillis();
            System.out.println("list1 = " + list1);
            System.out.println("例子1总耗时 = " + (endTime1 - startTime));
    
        }
    
       public String a1(){ 
           try { 
               TimeUnit.SECONDS.sleep(3);
               System.out.println("a1完成");
           } catch (InterruptedException e) { 
               e.printStackTrace();
           }
           return "yyyl";
        }
    
        public String a2(){ 
            try { 
                TimeUnit.SECONDS.sleep(4);
                System.out.println("a2完成");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return "yyy2";
        }

使用自定义的线程池替换  
//注 计算密集使用jdk的 ForkJoinPool线程池 IO密集使用自定义的线程池

/* * Copyright 2020 Wicrenet, Inc. All rights reserved. */
    package com.allianity.modules.report.service.impl;
    
    import com.allianity.common.learning.dto.DeptQuery;
    import com.allianity.common.learning.entity.SysDeptEntity;
    import com.allianity.modules.home.service.DeptService;
    import com.allianity.modules.report.controller.model.CourseGroupRateReportQuery;
    import com.allianity.modules.report.dto.CourseGroupRateBycgIdDTO;
    import com.allianity.modules.report.dto.CourseGroupRateReportByDeptIdDTO;
    import com.allianity.modules.report.dto.ReportUserDto;
    import com.allianity.modules.report.service.ReportCourseGroupUserService;
    import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
    import com.baomidou.mybatisplus.core.metadata.IPage;
    import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.BeanUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Service;
    
    import java.util.Comparator;
    import java.util.List;
    import java.util.Optional;
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.Executor;
    import java.util.stream.Collectors;
    
    /** * 【 任务完成率报表 】 * * @author YJX * Created on 2020/12/7 18:02 */
    @Service
    @Qualifier("base")
    public class CourseGroupRateReportServiceImpl extends AbsCourseGroupRateReportServiceImpl { 
        private static final Logger                       logger = LoggerFactory.getLogger(CourseGroupRateReportServiceImpl.class);
        @Autowired
        private              DeptService                  deptService;
        @Autowired
        private              ReportCourseGroupUserService reportCourseGroupUserService;
        @Autowired
        @Qualifier("asyncDataExecutor")
        private              Executor                     executor;
    
        // 1 组织维度统计
        @Override
        public Page<CourseGroupRateReportByDeptIdDTO> pageingByDeptId(CourseGroupRateReportQuery query) { 
            //当前查询组织
            SysDeptEntity one = this.deptService.getOne(new LambdaQueryWrapper<SysDeptEntity>()
                    .eq(SysDeptEntity::getDeptId, query.getDeptId())
            );
            //当前查询组织子节点
            IPage<SysDeptEntity> page = this.deptService.page(new Page<>(query.getPage(), query.getLimit()), new LambdaQueryWrapper<SysDeptEntity>()
                    .eq(SysDeptEntity::getParentId, one.getDeptId())
                    .eq(SysDeptEntity::getDelFlag, 0)
            );
            long l = System.currentTimeMillis();
            //注 计算密集使用jdk的 ForkJoinPool线程池 IO密集使用自定义的线程池
    // //并行流 jvm默认线程池
            List<CourseGroupRateReportByDeptIdDTO> reportDTOS1 = page.getRecords().parallelStream().map(item -> { 
                        try { 
                            Thread.sleep(2000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        CourseGroupRateReportByDeptIdDTO reportDTO = new CourseGroupRateReportByDeptIdDTO();
                        Optional.ofNullable(item.getDeptId()).ifPresent(reportDTO::setDeptId);
                        Optional.ofNullable(item.getLevelFlag()).ifPresent(reportDTO::setLevelFlag);
                        Optional.ofNullable(item.getName()).ifPresent(reportDTO::setDeptName);
                        //层级查询条件
                        DeptQuery deptQuery = this.deptService.setDeptid(item.getDeptId());
                        BeanUtils.copyProperties(deptQuery, query);
                        //在职人力
                        Long employeesStatistics = this.deptService.employeesStatisticsByDeptId(item.getDeptId());
                        reportDTO.setEmployeesStatistics(employeesStatistics);
                        //任务里面最大的参与人力
                        Long joinEmployees = this.reportCourseGroupUserService.joinEmployeesCountMax(query);
                        reportDTO.setJoinEmployees(joinEmployees);
                        //完成率
                        String avgCgPassRate = this.reportCourseGroupUserService.avgCgPassRate(query);
                        reportDTO.setCgPassRate(avgCgPassRate);
                        // 任务包两条数据
                        query.setPage(1);
                        query.setLimit(2);
                        query.setDeptId(item.getDeptId());
                        reportDTO.setCourseGroupRateBycgIdDtos(this.pageingByCgId(query).getRecords());
                        return reportDTO;
                    }
            ).collect(Collectors.toList());
            long l1 = System.currentTimeMillis();
            //异步多线程 自定义线程池 并每个多线程里面再次 异步并行塞数据
            List<CompletableFuture<CourseGroupRateReportByDeptIdDTO>> collect = page.getRecords().stream()
                    .map(item -> CompletableFuture
                            .supplyAsync(() -> { 
                                try { 
                                    Thread.sleep(2000);
                                } catch (InterruptedException e) { 
                                    e.printStackTrace();
                                }
                                CourseGroupRateReportByDeptIdDTO reportDTO = new CourseGroupRateReportByDeptIdDTO();
                                Optional.ofNullable(item.getDeptId()).ifPresent(reportDTO::setDeptId);
                                Optional.ofNullable(item.getLevelFlag()).ifPresent(reportDTO::setLevelFlag);
                                Optional.ofNullable(item.getName()).ifPresent(reportDTO::setDeptName);
                                //层级查询条件
                                DeptQuery deptQuery = this.deptService.setDeptid(item.getDeptId());
                                BeanUtils.copyProperties(deptQuery, query);
                                return reportDTO;
                            }, executor)
                            //在职人力
                            .thenCombineAsync(CompletableFuture.supplyAsync(() -> this.deptService.employeesStatisticsByDeptId(item.getDeptId())),
                                    (s1, s2) -> { 
                                        s1.setEmployeesStatistics(s2);
                                        return s1;
                                    }, executor)
                            //任务里面最大的参与人力
                            .thenCombineAsync(CompletableFuture.supplyAsync(() -> this.reportCourseGroupUserService.joinEmployeesCountMax(query)),
                                    (s1, s2) -> { 
                                        s1.setJoinEmployees(s2);
                                        return s1;
                                    }, executor)
                            //完成率
                            .thenCombineAsync(CompletableFuture.supplyAsync(() -> reportCourseGroupUserService.avgCgPassRate(query)),
                                    (s1, s2) -> { 
                                        s1.setCgPassRate(s2);
                                        return s1;
                                    }, executor)
                            // 任务包两条数据
                            .thenCombineAsync(CompletableFuture.supplyAsync(() -> { 
                                        query.setPage(1);
                                        query.setLimit(2);
                                        query.setDeptId(item.getDeptId());
                                        return this.pageingByCgId(query).getRecords();
                                    }),
                                    (s1, s2) -> { 
                                        s1.setCourseGroupRateBycgIdDtos(s2);
                                        return s1;
                                    }, executor)
                    ).collect(Collectors.toList());
            //这里需要分开写，不能把join操作写在上面，如果写在上面会导致每个线程变成串行
            List<CourseGroupRateReportByDeptIdDTO> reportDTOS = collect.stream().map(CompletableFuture::join)
                    .sorted(Comparator.comparing(CourseGroupRateReportByDeptIdDTO::getCgPassRate).reversed())
                    .collect(Collectors.toList());
            logger.debug("串行单线程-l1 = " + (l1 - l));
            logger.debug("异步多线程-l2 = " + (System.currentTimeMillis() - l1));
            Page<CourseGroupRateReportByDeptIdDTO> dtoPage = new Page<>();
            dtoPage.setRecords(reportDTOS);
            dtoPage.setTotal(page.getTotal());
            dtoPage.setSize(page.getSize());
            dtoPage.setCurrent(page.getCurrent());
            return dtoPage;
        }
    
        // 2 任务维度统计
        @Override
        public Page<CourseGroupRateBycgIdDTO> pageingByCgId(CourseGroupRateReportQuery query) { 
            //当前查询组织
            SysDeptEntity one = this.deptService.getOne(new LambdaQueryWrapper<SysDeptEntity>()
                    .eq(SysDeptEntity::getDeptId, query.getDeptId())
            );
            //层级查询条件
            DeptQuery deptQuery = this.deptService.setDeptid(one.getDeptId());
            BeanUtils.copyProperties(deptQuery, query);
            Page<CourseGroupRateBycgIdDTO> page = this.reportCourseGroupUserService.pageingByCgId(new Page<>(query.getPage(), query.getLimit()), query);
            page.getRecords().forEach(item -> { 
                Optional.ofNullable(one.getDeptId()).ifPresent(item::setDeptId);
                Optional.ofNullable(one.getLevelFlag()).ifPresent(item::setLevelFlag);
                Optional.ofNullable(one.getName()).ifPresent(item::setDeptName);
                //在职人力
                Long employeesStatistics = this.deptService.employeesStatisticsByDeptId(item.getDeptId());
                item.setEmployeesStatistics(employeesStatistics);
                item.setCgPassRate(StringUtils.isBlank(item.getCgPassRate()) ? "0.00" : String.format("%.2f", Double.parseDouble(item.getCgPassRate())) + "");
            });
            return page;
        }
    
        // 3 任务详情统计
        @Override
        public Page<ReportUserDto> pageingByCgIdAndUserId(CourseGroupRateReportQuery query) { 
            //当前查询组织
            SysDeptEntity one = this.deptService.getOne(new LambdaQueryWrapper<SysDeptEntity>()
                    .eq(SysDeptEntity::getDeptId, query.getDeptId())
            );
            //层级查询条件
            DeptQuery deptQuery = this.deptService.setDeptid(one.getDeptId());
            BeanUtils.copyProperties(deptQuery, query);
            return this.reportCourseGroupUserService.pageingByCgIdAndUserId(new Page<>(query.getPage(), query.getLimit()), query);
        }
    
    }

CompletableFuture<String> async = CompletableFuture.supplyAsync(() -> this.demoService.sayHello(name));
            CompletableFuture<String> async1 = CompletableFuture.supplyAsync(() -> this.demoService.sayHello(name));
            CompletableFuture.allOf(async, async1);
    // try { 
    // Thread.sleep(200);
    // } catch (InterruptedException e) { 
    // e.printStackTrace();
    // }
            String s = async.join();
            String s1 = async1.join();
    
    
    或者
            CompletableFuture<String> async3 = CompletableFuture.supplyAsync(() -> this.demoService.sayHello(name))
            .thenCombineAsync(CompletableFuture.supplyAsync(() -> this.demoService.sayHello(name)), (s1, s2) -> s1 + s2);
    		String s3 = async3.join();

并发对比

<table> 
 <thead> 
  <tr> 
   <th>描述</th> 
   <th>Future</th> 
   <th>FutureTask</th> 
   <th>CompletableFuture的比较</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>原理Future接口</td> 
   <td>Future Future</td> 
   <td>Runnable</td> 
   <td>CompletionStage</td> 
  </tr> 
  <tr> 
   <td>并发执行</td> 
   <td>支持</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>结果的顺序</td> 
   <td>提交顺序</td> 
   <td>未知</td> 
   <td>可指定顺序</td> 
  </tr> 
  <tr> 
   <td>异常捕捉</td> 
   <td>自己捕捉</td> 
   <td>自己捕捉</td> 
   <td>API中返回每个异常的生成</td> 
  </tr> 
 </tbody> 
</table>