Java-实现可监控线程池

爱被打了一巴掌 2022-02-27 15:16 254阅读 0赞

目标：创建线程池和一个监控线程，监控线程持有线程池引用，接着定时打印线程池相关信息，使用springboot创建demo。  
创建线程池接口：

/**
     * 通过设置的参数创建线程池
     * Created by wzx on 22/03/2019
     */
    public interface ITestThreadPoolExecutorService {
    
        ThreadPoolExecutor createThreadPoolExecutor();
    
        int getCorePoolSize();
    
        void setCorePoolSize(int corePoolSize);
    
        int getMaxPoolSize();
    
        void setMaxPoolSize(int maxPoolSize);
    
        long getKeepAliveTime();
    
        void setKeepAliveTime(long keepAliveTime);
    
    }

线程池实现类：

/**
     * Created by wzx on 22/03/2019
     */
    @Data
    public class TestThreadPoolExecutorService implements ITestThreadPoolExecutorService {
    
        private int corePoolSize;
        private int maxPoolSize;
        private long keepAliveTime;
        private int queueCapacity;
        private RejectedExecutionHandler rejectedExecutionHandler;
    
        @Override
        public ThreadPoolExecutor createThreadPoolExecutor() {
            return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<>(getQueueCapacity()),
                    getRejectedExecutionHandler());
        }
    
    }

线程池监控接口：

/**
     * 线程池监控接口
     * Created by wzx on 22/03/2019
     */
    public interface IThreadPoolMonitorService extends Runnable {
    
        void monitorThreadPool();
    
        ThreadPoolExecutor getExecutor();
    
        void setExecutor(ThreadPoolExecutor executor);
    
        long getMonitoringPeriod();
    
        void setMonitoringPeriod(long monitoringPeriod);
    
    }

线程池监控实现类：

/**
     * Created by wzx on 22/03/2019
     */
    @Data
    public class ThreadPoolMonitorService implements IThreadPoolMonitorService {
    
        private Logger logger = LoggerFactory.getLogger(ThreadPoolMonitorService.class);
    
        private ThreadPoolExecutor executor;
    
        private long monitoringPeriod;
    
        @Override
        public void monitorThreadPool() {
            String buffer = "CurrentPoolSize：" + executor.getPoolSize() +
                    " - CorePoolSize：" + executor.getCorePoolSize() +
                    " - MaximumPoolSize：" + executor.getMaximumPoolSize() +
                    " - ActiveTaskCount：" + executor.getActiveCount() +
                    " - CompletedTaskCount：" + executor.getCompletedTaskCount() +
                    " - TotalTaskCount：" + executor.getTaskCount() +
                    " - isTerminated：" + executor.isTerminated();
            logger.debug(buffer);
        }
    
        @Override
        public void run() {
            try {
                while (true) {
                    monitorThreadPool();
                    Thread.sleep(monitoringPeriod * 1000);
                }
            } catch (Exception e) {
                logger.error(e.getMessage());
            }
        }
    }

线程池和线程池监控服务启动类：

/**
     * 这个类维护了线程池和线程池监控服务
     * Created by wzx on 22/03/2019
     */
    @Component
    public class ThreadPoolExecutorStarter {
    
        private Logger logger = LoggerFactory.getLogger(ThreadPoolExecutorStarter.class);
    
        @Autowired
        private ITestThreadPoolExecutorService threadPoolExecutorService;
    
        @Autowired
        private IThreadPoolMonitorService monitorService;
    
        public void start() {
            ThreadPoolExecutor executor = threadPoolExecutorService.createThreadPoolExecutor();
            executor.allowCoreThreadTimeOut(true);
    
            monitorService.setExecutor(executor);
            monitorService.setMonitoringPeriod(3);
            Thread monitor = new Thread(monitorService);
            monitor.start();
    
            for (int i = 1; i <= 10; i++) {
                executor.execute(new TestTask("task" + i));
            }
    
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                logger.error(e.getMessage());
            }
    
            for (int i = 1; i <= 5; i++) {
                executor.execute(new TestTask("task" + i));
            }
    
            executor.shutdown();
        }
    
    
    }

任务拒绝执行策略：

/**
     * Created by wzx on 22/03/2019
     */
    public class TestRejectedExecutionHandler implements RejectedExecutionHandler {
    
        private Logger logger = LoggerFactory.getLogger(TestRejectedExecutionHandler.class);
    
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            logger.debug(r.toString() + ": has been rejected.");
        }
    
    }

普通任务类：

/**
     * Created by wzx on 22/03/2019
     */
    public class TestTask implements Runnable {
    
        private Logger logger = LoggerFactory.getLogger(TestTask.class);
    
        private String taskName;
    
        public TestTask(String taskName) {
            this.taskName = taskName;
        }
    
        @Override
        public void run() {
            try {
                logger.debug(taskName + " is started.");
                Thread.sleep(1000);
                logger.debug(taskName + " is completed.");
            } catch (InterruptedException e) {
                logger.error(taskName + " is not completed.");
                e.printStackTrace();
            }
        }
    
        public void setTaskName(String taskName) {
            this.taskName = taskName;
        }
    
        public String getTaskName() {
            return taskName;
        }
    
        @Override
        public String toString() {
            return getTaskName();
        }
    }

demo启动类：

@SpringBootApplication
    public class DemoApplication implements CommandLineRunner {
    
        public static void main(String[] args) {
            SpringApplication.run(DemoApplication.class, args);
        }
    
        @Bean
        public TestRejectedExecutionHandler rejectedExecutionHandler() {
            return new TestRejectedExecutionHandler();
        }
    
        @Bean
        public ITestThreadPoolExecutorService threadPoolExecutorService(TestRejectedExecutionHandler rejectedExecutionHandler) {
            TestThreadPoolExecutorService executor = new TestThreadPoolExecutorService();
            executor.setCorePoolSize(1);
            executor.setMaxPoolSize(3);
            executor.setQueueCapacity(3);
            executor.setKeepAliveTime(10);
            executor.setRejectedExecutionHandler(rejectedExecutionHandler);
            return executor;
        }
    
        @Bean
        public IThreadPoolMonitorService monitorService() {
            ThreadPoolMonitorService monitorService = new ThreadPoolMonitorService();
            monitorService.setMonitoringPeriod(3);
            return monitorService;
        }
    
        @Autowired
        private ThreadPoolExecutorStarter executorStarter;
    
        @Override
        public void run(String... args) throws Exception {
            //线程池监控测试
            executorStarter.start();
        }
    }

运行结果：

00:03:11.915 logback [Thread-10] DEBUG c.e.d.s.t.ThreadPoolMonitorService - CurrentPoolSize：0 - CorePoolSize：1 - MaximumPoolSize：3 - ActiveTaskCount：0 - CompletedTaskCount：0 - TotalTaskCount：0 - isTerminated：false
    00:03:11.916 logback [restartedMain] DEBUG c.e.d.s.t.TestRejectedExecutionHandler - task7: has been rejected.
    00:03:11.916 logback [pool-2-thread-3] DEBUG c.e.demo.service.threadpool.TestTask - task6 is started.
    00:03:11.916 logback [pool-2-thread-1] DEBUG c.e.demo.service.threadpool.TestTask - task1 is started.
    00:03:11.916 logback [pool-2-thread-2] DEBUG c.e.demo.service.threadpool.TestTask - task5 is started.
    00:03:11.916 logback [restartedMain] DEBUG c.e.d.s.t.TestRejectedExecutionHandler - task8: has been rejected.
    00:03:11.916 logback [restartedMain] DEBUG c.e.d.s.t.TestRejectedExecutionHandler - task9: has been rejected.
    00:03:11.916 logback [restartedMain] DEBUG c.e.d.s.t.TestRejectedExecutionHandler - task10: has been rejected.
    00:03:12.917 logback [pool-2-thread-3] DEBUG c.e.demo.service.threadpool.TestTask - task6 is completed.
    00:03:12.917 logback [pool-2-thread-1] DEBUG c.e.demo.service.threadpool.TestTask - task1 is completed.
    00:03:12.917 logback [pool-2-thread-2] DEBUG c.e.demo.service.threadpool.TestTask - task5 is completed.
    00:03:12.917 logback [pool-2-thread-3] DEBUG c.e.demo.service.threadpool.TestTask - task2 is started.
    00:03:12.917 logback [pool-2-thread-1] DEBUG c.e.demo.service.threadpool.TestTask - task3 is started.
    00:03:12.917 logback [pool-2-thread-2] DEBUG c.e.demo.service.threadpool.TestTask - task4 is started.
    00:03:13.918 logback [pool-2-thread-1] DEBUG c.e.demo.service.threadpool.TestTask - task3 is completed.
    00:03:13.918 logback [pool-2-thread-3] DEBUG c.e.demo.service.threadpool.TestTask - task2 is completed.
    00:03:13.918 logback [pool-2-thread-2] DEBUG c.e.demo.service.threadpool.TestTask - task4 is completed.
    00:03:14.915 logback [Thread-10] DEBUG c.e.d.s.t.ThreadPoolMonitorService - CurrentPoolSize：3 - CorePoolSize：1 - MaximumPoolSize：3 - ActiveTaskCount：0 - CompletedTaskCount：6 - TotalTaskCount：6 - isTerminated：false
    00:03:17.916 logback [Thread-10] DEBUG c.e.d.s.t.ThreadPoolMonitorService - CurrentPoolSize：3 - CorePoolSize：1 - MaximumPoolSize：3 - ActiveTaskCount：0 - CompletedTaskCount：6 - TotalTaskCount：6 - isTerminated：false