擴(kuò)展ThreadPoolExecutor

ThreadPoolExecutor是可擴(kuò)展的，通過查看源碼可以發(fā)現(xiàn)，它提供了幾個(gè)可以在子類化中改寫的方法：beforeExecute,afterExecute,terminated。源碼片段如下所示：

protected void beforeExecute(Thread t, Runnable r) { }  
protected void afterExecute(Runnable r, Throwable t) { }  
protected void terminated() { }  

可以注意到，這三個(gè)方法都是protected的空方法，擺明了是讓子類擴(kuò)展的嘛。

在執(zhí)行任務(wù)的線程中將調(diào)用beforeExecute和afterExecute等方法，在這些方法中還可以添加日志、計(jì)時(shí)、監(jiān)視或者統(tǒng)計(jì)信息收集的功能。無論任務(wù)是從run中正常返回，還是拋出一個(gè)異常而返回，afterExecute都會被調(diào)用。如果任務(wù)在完成后帶有一個(gè)Error，那么就不會調(diào)用afterExecute。如果beforeExecute拋出一個(gè)RuntimeException，那么任務(wù)將不被執(zhí)行，并且afterExecute也不會被調(diào)用。

在線程池完成關(guān)閉時(shí)調(diào)用terminated，也就是在所有任務(wù)都已經(jīng)完成并且所有工作者線程也已經(jīng)關(guān)閉后，terminated可以用來釋放Executor在其生命周期里分配的各種資源，此外還可以執(zhí)行發(fā)送通知、記錄日志或者收集finalize統(tǒng)計(jì)信息等操作。

下面就以給線程池添加統(tǒng)計(jì)信息為例：

public class TimingThreadPool extends ThreadPoolExecutor {
    private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();
    private final Logger log = Logger.getAnonymousLogger();
    private final AtomicLong numTasks = new AtomicLong();
    private final AtomicLong totalTime = new AtomicLong();

    public TimingThreadPool(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

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

    protected void afterExecute(Runnable r, Throwable t) {
        try {
            long endTime = System.nanoTime();
            long taskTime = endTime - startTime.get();
            numTasks.incrementAndGet();
            totalTime.addAndGet(taskTime);
            log.info(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();
        }
    }
}

可以看到TimingThreadPool重寫了父類的三個(gè)方法。

下面寫一個(gè)測試類，參考運(yùn)行效果：

public class CheckTimingThreadPool {
    public static void main(String[] args) {
        ThreadPoolExecutor exec = new TimingThreadPool(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
        exec.execute(new DoSomething(5));
        exec.execute(new DoSomething(4));
        exec.execute(new DoSomething(3));
        exec.execute(new DoSomething(2));
        exec.execute(new DoSomething(1));
        exec.shutdown();
    }
}

class DoSomething implements Runnable {
    private int sleepTime;

    public DoSomething(int sleepTime) {
        this.sleepTime = sleepTime;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " is running.");
        try {
            TimeUnit.SECONDS.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

運(yùn)行結(jié)果：

一月 20, 2018 5:34:51 下午 concurrent.TimingThreadPool beforeExecute
INFO: Thread Thread[pool-1-thread-5,5,main]: start concurrent.DoSomething@63425e17
pool-1-thread-5 is running.
一月 20, 2018 5:34:51 下午 concurrent.TimingThreadPool beforeExecute
INFO: Thread Thread[pool-1-thread-2,5,main]: start concurrent.DoSomething@156237e0
pool-1-thread-2 is running.
一月 20, 2018 5:34:51 下午 concurrent.TimingThreadPool beforeExecute
INFO: Thread Thread[pool-1-thread-3,5,main]: start concurrent.DoSomething@156237e0
pool-1-thread-3 is running.
一月 20, 2018 5:34:51 下午 concurrent.TimingThreadPool beforeExecute
INFO: Thread Thread[pool-1-thread-1,5,main]: start concurrent.DoSomething@156237e0
pool-1-thread-1 is running.
一月 20, 2018 5:34:51 下午 concurrent.TimingThreadPool beforeExecute
INFO: Thread Thread[pool-1-thread-4,5,main]: start concurrent.DoSomething@1a2f6aea
pool-1-thread-4 is running.
一月 20, 2018 5:34:52 下午 concurrent.TimingThreadPool afterExecute
INFO: Thread null: end concurrent.DoSomething@63425e17, time=1000557327ns
一月 20, 2018 5:34:53 下午 concurrent.TimingThreadPool afterExecute
INFO: Thread null: end concurrent.DoSomething@1a2f6aea, time=1999474354ns
一月 20, 2018 5:34:54 下午 concurrent.TimingThreadPool afterExecute
INFO: Thread null: end concurrent.DoSomething@156237e0, time=2999558243ns
一月 20, 2018 5:34:55 下午 concurrent.TimingThreadPool afterExecute
INFO: Thread null: end concurrent.DoSomething@156237e0, time=3999593049ns
一月 20, 2018 5:34:56 下午 concurrent.TimingThreadPool afterExecute
INFO: Thread null: end concurrent.DoSomething@156237e0, time=4999608220ns
一月 20, 2018 5:34:56 下午 concurrent.TimingThreadPool terminated
INFO: Terminated: avg time=2999758238ns

可以看到，在測試類CheckTimingThreadPool中通過execute了五個(gè)線程，然后分別對這五個(gè)線程進(jìn)行統(tǒng)計(jì)，最后統(tǒng)計(jì)出各個(gè)線程的耗時(shí)平均時(shí)間。