线程池ThreadPoolExecutor的一种扩展办法
概述? 在JAVA的世界里,如果想并行的执行一些任务,可以使用ThreadPoolExecutor。? 全宇宙的JAVA IT人士应该都知道ThreadPoolExecutor的执行流程:
这个执行流程有个小问题,就是当core线程无法应付请求的时候,会立刻将任务添加到队列中,如果队列非常长,而任务又非常多,那么将会有频繁的任务入队列和任务出队列的操作。 根据实际的压测发现,这种操作也是有一定消耗的。其实JAVA提供的SynchronousQueue队列是一个零长度的队列,任务都是直接由生产者递交给消费者,中间没有入队列的过程,可见JAVA API的设计者也是有考虑过入队列这种操作的开销。 另外,任务一多,立刻扔到队列里,而MAX线程又不干活,如果队列里面太多任务了,只有可怜的core线程在忙,也是会影响性能的。 当core线程无法应付请求的时候,能不能延后入队列这个操作呢? 让MAX线程尽快启动起来,帮忙处理任务。 也即是说,当core线程无法应付请求的时候,如果当前线程池中的线程数量还小于MAX线程数的时候,继续创建新的线程处理任务,一直到线程数量到达MAX后,才将任务插入到队列里 我们通过覆盖队列的offer方法来实现这个目标。 @Override public boolean offer(Runnable o) { int currentPoolThreadSize = executor.getPoolSize(); //如果线程池里的线程数量已经到达最大,将任务添加到队列中 if (currentPoolThreadSize == executor.getMaximumPoolSize()) { return super.offer(o); } 说明有空闲的线程,这个时候无需创建core线程之外的线程,而是把任务直接丢到队列里即可 if (executor.getSubmittedTaskCount() < currentPoolThreadSize) { .offer(o); } 如果线程池里的线程数量还没有到达最大,直接创建线程,而不是把任务丢到队列里面 if (currentPoolThreadSize <false; } .offer(o); } 注意其中的 .offer(o);
}
是表示core线程仍然能处理的来,同时又有空闲线程的情况,将任务插入到队列中。 如何判断线程池中有空闲线程呢? 可以使用一个计数器来实现,每当execute方法被执行的时候,计算器加1,当afterExecute被执行后,计数器减1。 public void execute(Runnable command) { submittedTaskCount.incrementAndGet(); 代码未完整,待补充。。。。。 } protected afterExecute(Runnable r,Throwable t) { submittedTaskCount.decrementAndGet(); } 这样,当 executor.getSubmittedTaskCount() < currentPoolThreadSize 的时候,说明有空闲线程。 ? ?完整代码? EnhancedThreadPoolExecutor类package executer; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; class EnhancedThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor { /** * 计数器,用于表示已经提交到队列里面的task的数量,这里task特指还未完成的task。 * 当task执行完后,submittedTaskCount会减1的。 */ private final AtomicInteger submittedTaskCount = new AtomicInteger(0); public EnhancedThreadPoolExecutor(int corePoolSize,int maximumPoolSize,1)">long keepAliveTime,TimeUnit unit,TaskQueue workQueue) { super(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,1)">new ThreadPoolExecutor.AbortPolicy()); workQueue.setExecutor(this); } * 覆盖父类的afterExecute方法,当task执行完成后,将计数器减1 */ @Override int getSubmittedTaskCount() { return submittedTaskCount.get(); } * 覆盖父类的execute方法,在任务开始执行之前,计数器加1。 try { .execute(command); } catch (RejectedExecutionException rx) { 当发生RejectedExecutionException,尝试再次将task丢到队列里面,如果还是发生RejectedExecutionException,则直接抛出异常。 BlockingQueue<Runnable> taskQueue = .getQueue(); if (taskQueue instanceof TaskQueue) { final TaskQueue queue = (TaskQueue)taskQueue; if (!queue.forceTaskIntoQueue(command)) { submittedTaskCount.decrementAndGet(); throw new RejectedExecutionException("队列已满"); } } else { submittedTaskCount.decrementAndGet(); throw rx; } } } } TaskQueuejava.util.concurrent.LinkedBlockingQueue; java.util.concurrent.RejectedExecutionException; class TaskQueue extends LinkedBlockingQueue<Runnable> { private EnhancedThreadPoolExecutor executor; public TaskQueue( capacity) { (capacity); } setExecutor(EnhancedThreadPoolExecutor exec) { executor = exec; } forceTaskIntoQueue(Runnable o) { if (executor.isShutdown()) { new RejectedExecutionException("Executor已经关闭了,不能将task添加到队列里面"); } .offer(o); } @Override offer(Runnable o) { executor.getPoolSize(); executor.getMaximumPoolSize()) { .offer(o); } currentPoolThreadSize) { .offer(o); } ; } .offer(o); } } TestExecuterjava.util.concurrent.TimeUnit; class TestExecuter { static final int CORE_SIZE = 5; int MAX_SIZE = 10long KEEP_ALIVE_TIME = 30int QUEUE_SIZE = 5static EnhancedThreadPoolExecutor executor = new EnhancedThreadPoolExecutor(CORE_SIZE,MAX_SIZE,KEEP_ALIVE_TIME,TimeUnit.SECONDS,1)"> TaskQueue(QUEUE_SIZE)); main(String[] args){ for (int i = 0; i < 15; i++) { executor.execute( Runnable() { @Override run() { { Thread.currentThread().sleep(1000); } (InterruptedException e) { e.printStackTrace(); } } }); System.out.println("线程池中现在的线程数目是:"+executor.getPoolSize()+",队列中正在等待执行的任务数量为:"+ executor.getQueue().size()); } } } 先运行一下代码,看看是否如何预期。直接执行TestExecuter类中的main方法,运行结果如下: 线程池中现在的线程数目是:1,队列中正在等待执行的任务数量为:0 线程池中现在的线程数目是:2,1)"> 线程池中现在的线程数目是:3,1)"> 线程池中现在的线程数目是:4,1)"> 线程池中现在的线程数目是:5,1)"> 线程池中现在的线程数目是:6,1)"> 线程池中现在的线程数目是:7,1)"> 线程池中现在的线程数目是:8,1)"> 线程池中现在的线程数目是:9,1)"> 线程池中现在的线程数目是:10,队列中正在等待执行的任务数量为:1 可以看到当线程数增加到core数量的时候,队列中是没有任务的。一直到线程数量增加到MAX数量,也即是10的时候,队列中才开始有任务。符合我们的预期。 如果我们注释掉TaskQueue类中的offer方法,也即是不覆盖队列的offer方法,那么运行结果如下: 可以看到当线程数增加到core数量的时候,队列中才开始有任务。符合我们的预期。
如果我们注释掉TaskQueue类中的offer方法,那么运行结果如下:
线程池中现在的线程数目是:1,队列中正在等待执行的任务数量为:5 在使用ThreadPoolExecutor的时候,如果发生了RejectedExecutionException,该如何处理?本文中的代码是采用了重新将任务尝试插入到队列中,如果还是失败则直接将reject异常抛出去。 @Override rx; } } } TaskQueue类提供了forceTaskIntoQueue方法,将任务插入到队列中。 还有另一种解决方案,就是使用另外一个线程池来执行任务,当第一个线程池抛出Reject异常时,catch住它,并使用第二个线程池处理任务。 ? 这篇文章在一个公众号里看到感觉写的很好,因为当是在做的项目正在使用线程池来解决问题,看过这篇文章后顺手就把自己项目里的线程池给扩展了一下。以前就想把这篇博客转过来收藏起来,因为一直联系不上作者就迟迟没有转,后来经过作者同意,就收入囊中了。 文章转自:https://blog.csdn.net/linsongbin1/article/details/78275283 作者:Sam哥哥 ? (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |