线程池监控：执行超时、等待超时；执行超时数量、等待超时数量

作者:匿名更新于： 2022-12-07 15:51:41

　　线程池 ThreadPoolExecutor 为了提供扩展，提供了两个方法 beforeExecute 和 afterExecute，每个任务执行前后都会调用这两个方法，相当于对线程任务的执行做了一个切面。

　　​监控线程池：执行超时、等待超时;执行超时数量、等待超时数量;

　　扩展线程池 ThreadPoolExecutor 的两个方法 beforeExecute 和 afterExecute

　　自定义Runnable 记录关键节点时间

　　关键时间节点参数：

• 　　任务创建(提交)时间：submitTime
• 　　任务开始执行时间：startExeTime
• 　　任务结束执行时间：endExeTime
• 　　任务在队列等待时间：任务开始执行时间 - 任务创建(提交)时间
• 　　任务执行总时间：任务结束执行时间 - 任务开始执行时间

　　源码分析

　　线程池 ThreadPoolExecutor 为了提供扩展，提供了两个方法 beforeExecute 和 afterExecute，每个任务执行前后都会调用这两个方法，相当于对线程任务的执行做了一个切面。

　　复制

　　1.  public class ThreadPoolExecutor extends AbstractExecutorService {

　　2.  /**

　　3.  * @param t 执行任务的线程

　　4.  * @param

　　5.  protected void beforeExecute(Thread t, Runnable r){ }

      6.

　　7.  /**

　　8.  * @param r 将要被执行的任务

　　9.  * @param

　　10.  protected void afterExecute(Runnable r, Throwable t){ }

　　11.  }

　　源码执行逻辑：

　　线程池扩展代码：

　　复制

　　1.  public class ThreadPoolExpandTest {

　　2.  // 定义线程池

　　3.  public static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(

　　4.  2,

　　5.  4,

　　6.  60,

　　7.  TimeUnit.SECONDS,

　　8.  new ArrayBlockingQueue<>(5),

　　9.  new ThreadPoolExecutor.DiscardOldestPolicy()

　　10.  ){

　　11.  @Override

　　12.  /**

　　13.  * @param t 执行任务的线程

　　14.  * @param

　　15.  protected void beforeExecute(Thread t, Runnable r){

　　16.  System.out.println("beforeExecute将要被执行");

　　17.  }

      18.

　　19.  /**

　　20.  * @param r 将要被执行的任务

　　21.  * @param

　　22.  @Override

　　23.  protected void afterExecute(Runnable r, Throwable t){

　　24.  System.out.println("afterExecute已经执行完毕");

　　25.  }

　　26.  };

　　27.  public static void main(String[] args){

　　28.  poolExecutor.execute(()->{

　　29.  System.out.println("任务执行");

　　30.  });

　　31.  }

　　32.  }

　　运行结果：

　　复制

　　1.  beforeExecute执行

　　2.  任务执行

　　3.  afterExecute执行

　　总结：从测试代码可以看出，通过扩展线程池参数可以进行任务执行的监控。

　　自定义Runnable

　　通过自定义Runnable，记录任务执行的一些时间：

• 　　任务创建(提交)时间
• 　　任务开始执行时间

　　复制

　　1.  public class DynamicRunnable implements Runnable{

　　2.  /**

　　3.  * runnable

　　4.  */

　　5.  private final Runnable runnable;

　　6.  /**

　　7.  * 任务创建(提交)时间

　　8.  */

　　9.  private final Long submitTime;

　　10.  /**

　　11.  * 任务开始执行时间

　　12.  */

　　13.  private Long startExeTime;

      14.

　　15.  public DynamicRunnable(Runnable runnable){

　　16.  this.runnable = runnable;

　　17.  submitTime = System.currentTimeMillis();

　　18.  }

      19.

　　20.  @Override

　　21.  public void run(){

　　22.  runnable.run();

　　23.  }

      24.

　　25.  public Long getSubmitTime(){

　　26.  return submitTime;

　　27.  }

      28.

　　29.  public void setStartExeTime(Long startExeTime){

　　30.  this.startExeTime = startExeTime;

　　31.  }

      32.

　　33.  public Long getStartExeTime(){

　　34.  return startExeTime;

　　35.  }

　　36.  }

　　继承线程池+自定义Runnable

　　核心参数：

　　复制

　　1.  /**

　　2.  * 执行超时，单位(毫秒)

　　3.  */

　　4.  private long runTimeout;

      5.

　　6.  /**

　　7.  * 等待超时，单位(毫秒)

　　8.  */

　　9.  private long queueTimeout;

     10.

　　11.  /**

　　12.  * 执行超时数量

　　13.  */

　　14.  private final AtomicInteger runTimeoutCount = new AtomicInteger();

      15.

　　16.  /**

　　17.  * 等待超时数量

　　18.  */

　　19.  private final AtomicInteger queueTimeoutCount = new AtomicInteger();

　　重写ThreadPoolExecutor方法：

　　复制

　　1.  @Override

　　2.  public void execute(Runnable command){

　　3.  if (runTimeout > 0 || queueTimeout > 0) {

　　4.  // 记录任务提交时间

　　5.  command = new DynamicRunnable(command);

　　6.  }

　　7.  super.execute(command);

　　8.  }

　　复制

　　1.  @Override

　　2.  protected void beforeExecute(Thread t, Runnable r){

　　3.  if (!(r instanceof DynamicRunnable)) {

　　4.  super.beforeExecute(t, r);

　　5.  return;

　　6.  }

　　7.  DynamicRunnable runnable = (DynamicRunnable) r;

　　8.  long currTime = System.currentTimeMillis();

　　9.  if (runTimeout > 0) {

　　10.  // 记录任务开始执行时间

　　11.  runnable.setStartExeTime(currTime);

　　12.  }

　　13.  if (queueTimeout > 0) {

　　14.  // 任务开始执行时间 - 任务创建(提交)时间

　　15.  long waitTime = currTime - runnable.getSubmitTime();

　　16.  if (waitTime > queueTimeout) {

　　17.  log.error("{} execute queue timeout waitTime: {}ms", this.getThreadPoolName(),waitTime);

　　18.  }

　　19.  }

　　20.  super.beforeExecute(t, r);

　　21.  }

　　复制

　　1.  @Override

　　2.  protected void afterExecute(Runnable r, Throwable t){

　　3.  if (runTimeout > 0) {

　　4.  DynamicRunnable runnable = (DynamicRunnable) r;

　　5.  // 任务执行总时间：任务结束执行时间 - 任务开始执行时间

　　6.  long runTime = System.currentTimeMillis() - runnable.getStartExeTime();

　　7.  if (runTime > runTimeout) {

　　8.  runTimeoutCount.incrementAndGet();

　　9.  log.error("{} execute, run timeout runTime: {}ms", this.getThreadPoolName(), runTime);

　　10.  }

　　11.  }

　　12.  super.afterExecute(r, t);

　　13.  }

　　来源： 今日头条

  　　>>>>>>点击进入编程语言专题