线程池的钩子方法

Posted 2021-02-28 cnxieyang

我们经常听说过钩子方法，高级模块会用到，本文结合线程池进行演示beforeExecute的钩子方法。见下实例代码，可控制线程池的运行和终止。

package threadPool;


import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 可中断的线程池演示
 * [e-mail]
 */
public class PauseThreadPool  extends ThreadPoolExecutor {
    /**
     * 是否已经中断你的标记
     */
    private boolean isPaused;
    private final ReentrantLock lock=new ReentrantLock();

    private Condition condition=lock.newCondition();
    public PauseThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

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

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

    public PauseThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        try {
            lock.lock();
            while(isPaused){
                condition.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void resume(){
        lock.lock();
        try {
            isPaused=false;
            condition.signalAll();
        }finally {
            lock.unlock();
        }
    }
    public void pause(){
        lock.lock();
        try{
            isPaused=true;
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
       PauseThreadPool pauseThreadPool= new PauseThreadPool(10,20,10L,TimeUnit.SECONDS,new LinkedBlockingDeque<>());
       Runnable runnable=new Runnable() {
           @Override
           public void run() {
               System.out.println("任务执行了");
               try {
                   Thread.sleep(20);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       };
       for (int i =0;i<100000;i++){
            pauseThreadPool.execute(runnable);
       }
       Thread.sleep(1000);
       pauseThreadPool.pause();
       System.out.println("该线程池被暂停了");

       Thread.sleep(2000);
       pauseThreadPool.resume();
       System.out.println("该线程池又恢复了");
    }

}