带有ThreadPoolExecutor Queque的PriorityBlockong不创建(动态)非核心线程

Question

在这里，看到我的优先级ThreadPoolExecutor -它工作很好，但问题是，它没有创建新的线程，如果实现了数量的с资源池е线程。

import java.util.Comparator;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by ngrigoriev on 4/24/14.
 */
public class PriorityExecutor extends ThreadPoolExecutor {

    public PriorityExecutor(int corePoolSize, int maxPoolSize, int quequSize) {
        super(corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS, new PriorityBlockingQueue<>(quequSize, new PriorityTaskComparator()));
    }

    public PriorityExecutor(int corePoolSize, int maxPoolSize, long time, TimeUnit unit, int quequSize) {
        super(corePoolSize, maxPoolSize, time, unit, new PriorityBlockingQueue<>(quequSize, new PriorityTaskComparator()));
    }

    public PriorityExecutor() {
        super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new PriorityBlockingQueue<>(11, new PriorityTaskComparator()));
    }

    public PriorityExecutor(final ThreadFactory threadFactory, int quequSize) {
        super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new PriorityBlockingQueue<>(quequSize, new PriorityTaskComparator()), threadFactory);
    }

    public PriorityExecutor(final RejectedExecutionHandler handler, int quequSize) {
        super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new PriorityBlockingQueue<>(quequSize, new PriorityTaskComparator()), handler);
    }

    public PriorityExecutor(final ThreadFactory threadFactory, final RejectedExecutionHandler handler, int quequSize) {
        super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new PriorityBlockingQueue<>(quequSize, new PriorityTaskComparator()), threadFactory,
                handler);
    }

    @Override
    public Future<?> submit(final Runnable task) {
        if (task == null)
            throw new NullPointerException();
        final RunnableFuture<Object> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null)
            throw new NullPointerException();
        final RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        final RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(final Callable<T> callable) {
        if (callable instanceof Important)
            return new PriorityTask<>(((Important) callable).getPriority(), callable);
        else
            return new PriorityTask<>(0, callable);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) {
        if (runnable instanceof Important)
            return new PriorityTask<>(((Important) runnable).getPriority(), runnable, value);
        else
            return new PriorityTask<>(0, runnable, value);
    }

    public interface Important {
        int getPriority();
    }

    private static final class PriorityTask<T> extends FutureTask<T> implements Comparable<PriorityTask<T>> {
        private final int priority;

        public PriorityTask(final int priority, final Callable<T> tCallable) {
            super(tCallable);

            this.priority = priority;
        }

        public PriorityTask(final int priority, final Runnable runnable, final T result) {
            super(runnable, result);

            this.priority = priority;
        }

        @Override
        public int compareTo(final PriorityTask<T> o) {
            final long diff = o.priority - priority;
            return 0 == diff ? 0 : 0 > diff ? -1 : 1;
        }
    }

    private static class PriorityTaskComparator implements Comparator<Runnable> {
        @Override
        public int compare(final Runnable left, final Runnable right) {
            return ((PriorityTask) left).compareTo((PriorityTask) right);
        }
    }
}

在调试过程中，我发现在第1368行的方法execute中，我们给出了一个条件来创建一个非核心工作人员，但是这个条件永远不成立--这个方法(workerCountOf(recheck))不能在没有调试字节码的情况下进行调试。

    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
       1368 **else if (workerCountOf(recheck) == 0)**
            addWorker(null, false);
    }

这其实是个答案

池只在队列无法接受任务时创建非核心线程。队列总是接受任务，因此不会创建更多的线程。

谢谢，但我是否可以更改线程池策略以按某些条件添加新的非核心线程？？例如，通过等待核心线程状态。

Answer 1

您将需要提供一个自定义BlockingQueue和RejectedExecutionHandler实现，它包装了您的优先级队列。此自定义队列将需要对ThreadPoolExecutor的引用，并根据ThreadPoolExecutor的当前活动计数选择是否接受“提供”的可运行实例。它还必须是ThreadPoolExecutor的拒绝执行处理程序，以便如果同时提供多个对象，导致超出最大大小，则作业仍将添加到优先级队列中。

另一种选择(这是/大得多/更简单)是简单地将“核心”大小设置为您希望允许的最大值和允许核心线程超时。