多线程 EagerThreadPool（dubbo）程池的实现

本文主要是介绍多线程 EagerThreadPool（dubbo）程池的实现，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

EagerThreadPool 种线程池的实现
任务处理优先线程池
ThreadPoolExecutor普通线程执行器。当线程池核心线程达到阈值时新任务放入队列。当队列已满开启新线程处理。当前线程数达到最大线程数时执行拒绝策略。

EagerThreadPoolExecutor自定义线程执行器。当线程池核心线程达到阈值时，新任务不会放入队列而是开启新线程进行处理（要求当前线程数没有超过最大线程数）。当前线程数达到最大线程数时任务放入队列。队列已满执行拒绝策略
源码如下：

public class EagerThreadPool implements ThreadPool {@Overridepublic Executor getExecutor(URL url) {// 线程名String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME);//核心线程数int cores = url.getParameter(CORE_THREADS_KEY, DEFAULT_CORE_THREADS);// 最大线程数默认Int最大值int threads = url.getParameter(THREADS_KEY, Integer.MAX_VALUE);// 队列容量默认0int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);// 线程空闲多少时间被回收默认1分钟int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE);// init queue and executor// 初始化自定义线程池和队列TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,threads,alive,TimeUnit.MILLISECONDS,taskQueue,new NamedInternalThreadFactory(name, true),new AbortPolicyWithReport(name, url));taskQueue.setExecutor(executor);return executor;}
}public class EagerThreadPoolExecutor extends ThreadPoolExecutor {// 提交任务个数private final AtomicInteger submittedTaskCount = new AtomicInteger(0);@Overridepublic void execute(Runnable command) {if (command == null) {throw new NullPointerException();}// 任务数自增submittedTaskCount.incrementAndGet();try {// 调用父类方法执行线程任务super.execute(command);}// 抛出拒绝异常catch (RejectedExecutionException rx) {final TaskQueue queue = (TaskQueue) super.getQueue();try {// 任务重新放入队列if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {// 任务重新放入队列失败抛出异常submittedTaskCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.", rx);}} catch (InterruptedException x) {submittedTaskCount.decrementAndGet();throw new RejectedExecutionException(x);}} catch (Throwable t) {submittedTaskCount.decrementAndGet();throw t;}}
}
public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {@Overridepublic boolean offer(Runnable runnable) {if (executor == null) {throw new RejectedExecutionException("The task queue does not have executor");}// 当前线程数int currentPoolThreadSize = executor.getPoolSize();// 任务数 < 当前线程数表示存在空闲worker线程则任务放入队列等待worker线程处理if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {return super.offer(runnable);}// 当前线程数 < 最大线程数返回false表示创建worker线程if (currentPoolThreadSize < executor.getMaximumPoolSize()) {return false;}// 当前线程数 > 最大线程数任务放入队列return super.offer(runnable);}public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if (executor.isShutdown()) {throw new RejectedExecutionException("Executor is shutdown!");}// 任务重试放入队列return super.offer(o, timeout, unit);}
}