扩展ThreadPoolExecutor

目录

• 1、什么是扩展ThreadPoolExecutor
• 2、哪些情况下可以进行扩展？
• 3、如何进行扩展？
• 4、注意事项
• 5、一个简单的示例

1、什么是扩展ThreadPoolExecutor

ThreadPoolExecutor是可扩展的，它提供了几个可以在子类中改写的方法：beforeExecute、afterExecute、terminated。这些方法可以用于扩展ThreadPoolExecutor的行为。

2、哪些情况下可以进行扩展？

在执行任务的线程中会调用beforeExecute和afterExecute等方法，在这些方法中可以添加日志、计时、监视或统计信息收集等功能。

3、如何进行扩展？

新建一个自己的线程池类，继承ThreadPoolExecutor。按自己的需要重写方法。

4、注意事项

1. 无论任务是从run中正常返回，还是抛出一个异常而返回，afterExecute都会被调用。如果任务在完成后带有一个Error，那么就不会调用afterExecute。还没到afterExecute就出现错误了，afterExecute就不调用了。
2. 如果beforeExecute抛出一个RuntimeException，那么任务将不被执行，并且afterExecute也不会被调用。beforeExecute就出现运行时异常，就不继续了。
3. 在线程完成关闭操作时调用terminated，也就是在所有任务都已经完成并且所有工作者线程也已经关闭后。terminated可以用来释放Executor在其生命周期里分配的各种资源，此外还可以执行发送通知、记录日志或者收集finalize统计信息等操作。

5、一个简单的示例

/**
 * 《Java并发编程实战》程序清单8-9
 * 扩展ThreadPoolExecutor，实现一个自定义的线程池，
 * 它通过beforeExecute、afterExecutor和terminated等方法添加日志记录和统计信息收集。
 *
 */
public class _08_09_TimingThreadPool extends ThreadPoolExecutor {

    public _08_09_TimingThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                BlockingQueue<Runnable> workQueue) {
        super(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
    }

    private final ThreadLocal<Long> startTime = new ThreadLocal<>();
    private final Logger log = Logger.getLogger("_08_09_TimingThreadPool");

    private final AtomicLong numTasks = new AtomicLong();

    private final AtomicLong totalTime = new AtomicLong();

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        log.fine(String.format("Thread %s: start %s", t, r));
        startTime.set(System.nanoTime());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        try {
            long endTime = System.nanoTime();
            long taskTime = endTime - startTime.get();
            numTasks.incrementAndGet();
            totalTime.addAndGet(taskTime);
            log.fine(String.format("Thread %s: end %s, time=%dns", t, r, taskTime));
        } finally {
            super.afterExecute(r, t);
        }
    }

    protected void terminated() {
        try {
            log.info(String.format("Terminated: avg time=%dns", totalTime.get() / numTasks.get()));
        } finally {
            super.terminated();
        }
        Executors.newFixedThreadPool(10);
    }
}