Java并发之并发优先线程池

Java原生线程池在提交任务时会优先将线程数扩展到 CoreSize，然后会将任务入队，在队列塞满后会尝试继续扩展线程数到MaxSize。这种方式适合cpu密集型任务，而且任务时间不宜过长，否则会造成队列里面任务的堆积。

对于 RPC 通信场景的 IO 密集型任务，这种调度方式就不太合适。更适合并发优先的调度策略，即优先扩展线程数到 MaxSize。然后再尝试入队。要是实现上面的调度策略需要基于 JDK 原生线程池做一下调整。

任务队列

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TaskQueue
        extends LinkedBlockingQueue<Runnable>
{
    private volatile ThreadPoolExecutor executor;

    public void setExecutor(ThreadPoolExecutor executor)
    {
        this.executor = executor;
    }

    @Override
    public boolean offer(Runnable runnable)
    {
        if (executor == null) {
            throw new RejectedExecutionException("The task queue does not have executor!");
        }

        int currentPoolThreadSize = executor.getPoolSize();
        // have free worker. put task into queue to let the worker deal with task.
        if (executor.getActiveCount() < currentPoolThreadSize) {
            return super.offer(runnable);
        }

        // return false to let executor create new worker.
        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
            return false;
        }

        // currentPoolThreadSize >= max
        return super.offer(runnable);
    }

    /**
     * Forcefully enqueue the rejected task.
     *
     * @param runnable task
     * @return offer success or not
     * @throws RejectedExecutionException if executor is terminated.
     */
    public boolean forceOffer(Runnable runnable, long timeout, TimeUnit unit)
            throws InterruptedException
    {
        if (executor == null || executor.isShutdown()) {
            throw new RejectedExecutionException("Executor is shutdown!");
        }
        return super.offer(runnable, timeout, unit);
    }
}

线程池

/**
 * 并发优先的线程池
 */
public class EagerThreadPool
        extends ThreadPoolExecutor
{
    public EagerThreadPool(int corePoolSize,
            int maximumPoolSize,
            long keepAliveTime,
            TimeUnit unit,
            BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory,
            RejectedExecutionHandler handler)
    {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    public void execute(Runnable command)
    {
        if (command == null) {
            throw new NullPointerException();
        }
        executeInternal(command, 0, TimeUnit.MILLISECONDS);
    }

    public void executeInternal(Runnable command, long timeout, TimeUnit unit)
    {
        try {
            super.execute(command);
        }
        catch (RejectedExecutionException rx) {
            if (getQueue() instanceof TaskQueue) {
                // If the Executor is close to maximum pool size, concurrent
                // calls to execute() may result in some tasks being rejected rather than queued.
                // If this happens, add them to the queue.
                final TaskQueue queue = (TaskQueue) getQueue();
                try {
                    if (!queue.forceOffer(command, timeout, unit)) {
                        throw new RejectedExecutionException("Queue capacity is full.", rx);
                    }
                }
                catch (InterruptedException x) {
                    throw new RejectedExecutionException(x);
                }
            }
            else {
                throw rx;
            }
        }
    }
}