浅谈可扩展线程池

public class TaskQueue<R extends Runnable>
    extends LinkedBlockingQueue<Runnable> {
    
    // ...
    
    @Override
    public boolean offer(final Runnable runnable) {
        // 未设置线程池的时候无法获取已提交的数量，抛出异常
        if (executor == null) {
            throw new RejectedExecutionException(
                "The task queue does not have executor!"
            );
        }

        final int currentPoolThreadSize = executor.getPoolSize();
        // 已提交的任务数量少于线程池当前启动的线程数量，则直接添加到工作队列中
        if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
            return super.offer(runnable);
        }

        // 判断当前线程数量是否达到最大线程数量，如果未达到，则返回 false，让线程池优先新建线程
        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
            return false;
        }

        // 当当前线程数量达到最大线程数量的时候，此时将任务添加到任务队列中
        return super.offer(runnable);
    }

    public boolean retryOffer(
        final Runnable o,
        final long timeout,
        final TimeUnit unit
    )
        throws InterruptedException {
        if (executor.isShutdown()) {
            throw new RejectedExecutionException("Executor is shutdown!");
        }
        // 重试插入
        return super.offer(o, timeout, unit);
    }
}

public class ThreadPoolExecutor
    extends java.util.concurrent.ThreadPoolExecutor {
    private final AtomicInteger submittedTaskCount = new AtomicInteger(0);

    // ...

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

    @Override
    protected void afterExecute(final Runnable r, final Throwable t) {
        // 完成任务后将提交的数量递减一，代表已经完成一个任务
        this.submittedTaskCount.decrementAndGet();
    }

    @Override
    @SuppressWarnings("rawtypes")
    public void execute(final Runnable command) {
        if (command == null) {
            throw new NullPointerException();
        }
        // 提交任务的时候递增一，代表有新的任务加入队列
        submittedTaskCount.incrementAndGet();
        try {
            // 实际执行任务
            super.execute(command);
        } catch (final 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
                    );
                }
                // else 插入成功，说明工作队列未满，只是未达到最大线程数，线程创建达到要求的时候就会执行
            } catch (final InterruptedException x) {
                submittedTaskCount.decrementAndGet();
                throw new RejectedExecutionException(x);
            }
        } catch (final Throwable t) {
            // 出现其他异常，则抛出异常
            submittedTaskCount.decrementAndGet();
            throw t;
        }
    }
}