Java主线程等待线程池所有任务完成--线程池的封装

墨蓝 2022-06-14 08:17 475阅读 0赞

> 用线程池编写多线程程序时，当所有任务完成时，要做一些统计的工作。而统计工作必须要在所有任务完成才能做。所以要让主线程等待所有任务完成。可以使用**ThreadPoolExecutor.awaitTermination(long timeout, TimeUnit unit)**。

### 封装的线程池 ###

public interface BaseExecutor { 
    
        /** * @Title: initial * @Description: 初始化组件 * @return: void */
        void initial();
    
        /** * @Title: shutdown * @Description: 关机 * @return: void */
        void shutdown();
    
        /** * * @Title: startup * @Description: 开启线程池 * @return: void */
        void startup();
    
        /** * @Title: execute * @param callable * @throws Exception * @return: void */
        void execute(Callable<T> callable) throws Exception;
        /** * * @Title: setThreadSize * @Description: 设置线程池里面核心线程的大小 * @param thread_size * @return: void */
        public void setThreadSize(int thread_size);
    }

### 计算线程池的实现 ###

public class ComputerExecutor implements BaseExecutor { 
    
        private ThreadPoolExecutor executor = null;
        private int DEFAULT_CORE_SIZE = 8; // 默认核心线程数
        private final int MAXIMUMPOOLSIZE = 50; // 默认最大线程数
        private long keepAliveTime = 30000; // 当线程池中的线程数量大于
                                            // corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。
    
        @Override
        public void initial() {
            executor = new ThreadPoolExecutor(DEFAULT_CORE_SIZE, MAXIMUMPOOLSIZE, keepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
    
        }
    
        @Override
        public void shutdown() {
            executor.shutdown();
            try {
                // 当线程池中的所有任务执行完毕时,就会关闭线程池
                while (!executor.awaitTermination(2, TimeUnit.SECONDS))
                    ;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void startup() {
            this.initial();
        }
    
        @Override
        public void execute(Callable<T> callable) throws Exception {
            process(callable);
        }
    
        private void process(Callable<T> callable) {
            FutureTask<T> future = new FutureTask<T>((Callable<T>) callable);
            executor.execute(future);
        }
    
        public void setThreadSize(int thread_size) {
            this.DEFAULT_CORE_SIZE = thread_size;
        }
    }

### 计算任务的封装 ###

public class CustomizeIndicatorComputerCallable<T> implements Callable<T>{
    
        @Autowired
        private CorplutionIndicatorReportFacade corplutionIndicatorReportFacade;
    
        private static Logger logger = Logger.getLogger(CustomizeIndicatorComputerCallable.class);
    
        private int bound_size = 10; //信号量--并发队列大小 
    
        private Semaphore semaphore = null;
    
        private Object[] args;
    
        public CustomizeIndicatorComputerCallable(
                int bound, Semaphore semaphore,  Object ...args) {
            this.bound_size = bound;
            this.semaphore = semaphore;
            this.args = args;
            @SuppressWarnings("resource")
            ApplicationContext context = new ClassPathXmlApplicationContext("classpath*:/META-INF/spring/*.xml");
            corplutionIndicatorReportFacade = (CorplutionIndicatorReportFacade)context.getBean("corplutionIndicatorReportFacade");
        }
    
        public T call() throws Exception {
            semaphore.acquire();
            try {
                logger.info("线程 ["+Thread.currentThread().getId()+"]线进入, 当前并发数 ["+ (bound_size - semaphore.availablePermits()) +"]");
                CorplutionIndicatorReportDO corplutionIndicatorReportDO = (CorplutionIndicatorReportDO)args[0];
                CorplutionCostomizeIndicator costomizeIndicator = (CorplutionCostomizeIndicator)args[1];
                corplutionIndicatorReportFacade.computerCustomizeIndicator(corplutionIndicatorReportDO , costomizeIndicator);
            } catch (Exception e) {
                logger.info("线程池处理失败", e);
                throw e;
            }
            finally{
                semaphore.release();
                logger.info("线程 ["+Thread.currentThread().getId()+"] 离开,当前并发数 [ "+ (bound_size - semaphore.availablePermits()) +"]");
            }
            return null;
        }
    
    }