import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

/**

* Created by yehan .

*  代码和思路主要来自于：

*

* tomcat :

*        org.apache.catalina.core.StandardThreadExecutor

*

* java.util.concurrent

* threadPoolExecutor execute执行策略：        优先offer到queue，queue满后再扩充线程到maxThread，如果已经到了maxThread就reject

*                              比较适合于CPU密集型应用（比如runnable内部执行的操作都在JVM内部，memory copy, or compute等等）

*

* StandardThreadExecutor execute执行策略： 优先扩充线程到maxThread，再offer到queue，如果满了就reject

*                          比较适合于业务处理需要远程资源的场景

*

* @author yehan

*

*/

public class StandardThreadExecutorextends ThreadPoolExecutor {

public static final int DEFAULT_MIN_THREADS =20;

    public static final int DEFAULT_MAX_THREADS =200;

    /**1 minutes*/

    public static final int DEFAULT_MAX_IDLE_TIME =60 *1000;

    /**正在处理的任务数*/

    protected AtomicIntegersubmittedTasksCount;

    /**最大允许同时处理的任务数*/

    private int maxSubmittedTaskCount;

    public StandardThreadExecutor() {

this(DEFAULT_MIN_THREADS, DEFAULT_MAX_THREADS);

    }

public StandardThreadExecutor(int coreThread, int maxThreads) {

this(coreThread, maxThreads, maxThreads);

    }

public StandardThreadExecutor(int coreThread, int maxThreads, long keepAliveTime, TimeUnit unit) {

this(coreThread, maxThreads, keepAliveTime, unit, maxThreads);

    }

public StandardThreadExecutor(int coreThreads, int maxThreads, int queueCapacity) {

this(coreThreads, maxThreads, queueCapacity, Executors.defaultThreadFactory());

    }

public StandardThreadExecutor(int coreThreads, int maxThreads, int queueCapacity, ThreadFactory threadFactory) {

this(coreThreads, maxThreads, DEFAULT_MAX_IDLE_TIME, TimeUnit.MILLISECONDS, queueCapacity, threadFactory);

    }

public StandardThreadExecutor(int coreThreads, int maxThreads, long keepAliveTime, TimeUnit unit, int queueCapacity) {

this(coreThreads, maxThreads, keepAliveTime, unit, queueCapacity, Executors.defaultThreadFactory());

    }

public StandardThreadExecutor(int coreThreads, int maxThreads, long keepAliveTime, TimeUnit unit,

                                  int queueCapacity, ThreadFactory threadFactory) {

this(coreThreads, maxThreads, keepAliveTime, unit, queueCapacity, threadFactory, new AbortPolicy());

    }

public StandardThreadExecutor(int coreThreads, int maxThreads, long keepAliveTime, TimeUnit unit,

                                  int queueCapacity, ThreadFactory threadFactory, RejectedExecutionHandler handler) {

super(coreThreads, maxThreads, keepAliveTime, unit, new ExecutorQueue(), threadFactory, handler);

        ((ExecutorQueue) getQueue()).setStandardThreadExecutor(this);

        submittedTasksCount =new AtomicInteger(0);

        // 最大并发任务限制： 队列buffer数 + 最大线程数

        maxSubmittedTaskCount = queueCapacity + maxThreads;

    }

@Override

    public void execute(Runnable command) {

int count =submittedTasksCount.incrementAndGet();

        // 超过最大的并发任务限制，进行 reject

// 依赖的LinkedTransferQueue没有长度限制，因此这里进行控制

        if (count >maxSubmittedTaskCount) {

submittedTasksCount.decrementAndGet();

            getRejectedExecutionHandler().rejectedExecution(command, this);

        }

try {

super.execute(command);

        }catch (RejectedExecutionException rx) {

// there could have been contention around the queue

            if (!((ExecutorQueue) getQueue()).force(command)) {

submittedTasksCount.decrementAndGet();

                getRejectedExecutionHandler().rejectedExecution(command, this);

            }

}

}

public int getSubmittedTasksCount() {

return this.submittedTasksCount.get();

    }

public int getMaxSubmittedTaskCount() {

return maxSubmittedTaskCount;

    }

@Override

    protected void afterExecute(Runnable r, Throwable t) {

submittedTasksCount.decrementAndGet();

    }

}

/**

* LinkedTransferQueue 能保证更高性能，相比与LinkedBlockingQueue有明显提升

*

*

*        1) 不过LinkedTransferQueue的缺点是没有队列长度控制，需要在外层协助控制

*

*

* @author yehan

*

*/

class ExecutorQueueextends LinkedTransferQueue {

private static final long serialVersionUID = -265236426751004839L;

    StandardThreadExecutorthreadPoolExecutor;

    public ExecutorQueue() {

super();

    }

public void setStandardThreadExecutor(StandardThreadExecutor threadPoolExecutor) {

this.threadPoolExecutor = threadPoolExecutor;

    }

/** 注：代码来源于 tomcat*/

    public boolean force(Runnable o) {

if (threadPoolExecutor.isShutdown()) {

throw new RejectedExecutionException("Executor not running, can't force a command into the queue");

        }

// forces the item onto the queue, to be used if the task is rejected

        return super.offer(o);

    }

/**

*注：tomcat的代码进行一些小变更

*

*/

    @Override

    public boolean offer(Runnable o) {

int poolSize =threadPoolExecutor.getPoolSize();

        // we are maxed out on threads, simply queue the object

        if (poolSize ==threadPoolExecutor.getMaximumPoolSize()) {

return super.offer(o);

        }

// we have idle threads, just add it to the queue

// note that we don't use getActiveCount(), see BZ 49730

        if (threadPoolExecutor.getSubmittedTasksCount() <= poolSize) {

return super.offer(o);

        }

// if we have less threads than maximum force creation of a new

// thread

        if (poolSize

return false;

        }

// if we reached here, we need to add it to the queue

        return super.offer(o);

    }

}