自定义的异步线程池组件

AsyncThreadExecutor.java

public class AsyncThreadExecutor implements AutoCloseable {

    private static final int DEFAULT_QUEUE_SIZE = 1000;

    private static final int DEFAULT_POOL_SIZE = 10;

    @Setter
    private int queueSize = DEFAULT_QUEUE_SIZE;

    @Setter
    private int poolSize = DEFAULT_POOL_SIZE;

    /**
     * 用于周期性监控线程池的运行状态
     */
    private final ScheduledExecutorService scheduledExecutorService =
        Executors.newSingleThreadScheduledExecutor(new BasicThreadFactory.Builder().namingPattern("async thread executor monitor").build());

    /**
     * 自定义异步线程池
     * （1）任务队列使用有界队列
     * （2）自定义拒绝策略
     */
    private final ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue(queueSize),
                               new BasicThreadFactory.Builder().namingPattern("async-thread-%d").build(),
                               (r, executor) -> log.error("the async executor pool is full!!"));
    private final ExecutorService executorService = threadPoolExecutor;

    @PostConstruct
    public void init() {
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            /**
             * 线程池需要执行的任务数
             */
            long taskCount = threadPoolExecutor.getTaskCount();
            /**
             * 线程池在运行过程中已完成的任务数
             */
            long completedTaskCount = threadPoolExecutor.getCompletedTaskCount();
            /**
             * 曾经创建过的最大线程数
             */
            long largestPoolSize = threadPoolExecutor.getLargestPoolSize();
            /**
             * 线程池里的线程数量
             */
            long poolSize = threadPoolExecutor.getPoolSize();
            /**
             * 线程池里活跃的线程数量
             */
            long activeCount = threadPoolExecutor.getActiveCount();

            log.info("async-executor monitor. taskCount:{}, completedTaskCount:{}, largestPoolSize:{}, poolSize:{}, activeCount:{}",
                     taskCount, completedTaskCount, largestPoolSize, poolSize, activeCount);
        }, 0, 10, TimeUnit.MINUTES);
    }

    public void execute(Runnable task) {
        executorService.execute(task);
    }

    @Override
    public void close() throws Exception {
        executorService.shutdown();
    }
}