概述

在JAVA的世界里,假如想并行的执行一些任务,可以利用ThreadPoolExecutor。
大部门环境下直接利用ThreadPoolExecutor就可以满意要求了,可是在某些场景下,好比瞬时大流量的,为了提高响应和吞吐量,最好照旧扩展一下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) {
        return super.offer(o);
    }

    //假如线程池里的线程数量还没有达到最大,直接建设线程,而不是把任务丢到行列内里
    if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
        return false;
    }

    return super.offer(o);
}

留意个中的

if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
        return super.offer(o);
}

是暗示core线程仍然能处理惩罚的来,同时又有空闲线程的环境,将任务插入到行列中。 如何判定线程池中有空闲线程呢？ 可以利用一个计数器来实现,每当execute要领被执行的时候,计较器加1,当afterExecute被执行后,计数器减1.

    @Override
    public void execute(Runnable command) {
        submittedTaskCount.incrementAndGet();
        //代码未完整,待增补。。。。。
    }

 @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedTaskCount.decrementAndGet();
    }

这样,当

executor.getSubmittedTaskCount() < currentPoolThreadSize

的时候,说明有空闲线程。

完整代码

EnhancedThreadPoolExecutor类

package executer;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public 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, long keepAliveTime, TimeUnit unit, TaskQueue workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new ThreadPoolExecutor.AbortPolicy());
        workQueue.setExecutor(this);
    }

    /**
     * 包围父类的afterExecute要领,当task执行完成后,将计数器减1
     */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedTaskCount.decrementAndGet();
    }


    public int getSubmittedTaskCount() {
        return submittedTaskCount.get();
    }


    /**
     * 包围父类的execute要领,在任务开始执行之前,计数器加1。
     */
    @Override
    public void execute(Runnable command) {
        submittedTaskCount.incrementAndGet();
        try {
            super.execute(command);
        } catch (RejectedExecutionException rx) {
            //当产生RejectedExecutionException,实验再次将task丢到行列内里,假如照旧产生RejectedExecutionException,则直接抛出异常。
            BlockingQueue<Runnable> taskQueue = super.getQueue();
            if (taskQueue instanceof TaskQueue) {
                final TaskQueue queue = (TaskQueue)taskQueue;
                if (!queue.forceTaskIntoQueue(command)) {
                    submittedTaskCount.decrementAndGet();
                    throw new RejectedExecutionException("行列已满");
                }
            } else {
                submittedTaskCount.decrementAndGet();
                throw rx;
            }
        }
    }
}