在 ScheduledThreadPoolExecutor 中使用 PriorityBlockingQueue 和 Comparator

Question

首先：我已经阅读了以下两个问题及其可能的解决方案：

• ScheduledThreadPoolExecutors 和自定义队列
• Java Executors：如何设置任务优先级？

我遇到的困境是我想使用自定义队列BlockingQueue，或者更确切地说，使用不同但特定的队列，即PriorityBlockingQueue使用自定义队列Comparator使用按优先级对队列进行排序的自定义

它ThreadPoolExecutor确实支持在其构造函数中自定义队列，但它不实现接口中的方法ScheduledExecutorService。所以我去找了子类ScheduledThreadPoolExecutor，但它不支持自定义队列并使用DelayedWorkQueue。

问题：

• 我无法扩展，ScheduledThreadPoolExecutor因为为我自己的类创建构造函数不会做任何事情，因为ScheduledThreadPoolExecutor不接受自定义队列作为参数。
• 我无法复制类的内容ThreadPoolExecutor和实现，ScheduledThreadPoolExecutor因为它使用了许多没有修饰符声明的方法（例如canRunInCurrentState(boolean periodic)，此调用调用的所有方法），这不允许我访问该方法，因为即使它的子类ThreadPoolExecutor，它不在同一个包中。

我当前的实现如下所示：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.croemheld.tasks.PriorityTaskComparator;

public class ScheduledPriorityThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {

    private static final int INITIAL_QUEUE_SIZE = 10;

    public ScheduledPriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
            TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, 
            new PriorityBlockingQueue<Runnable>(INITIAL_QUEUE_SIZE, new PriorityTaskComparator()));
    }

    public ScheduledPriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
            TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, 
            new PriorityBlockingQueue<Runnable>(INITIAL_QUEUE_SIZE, new PriorityTaskComparator()), handler);
    }

    public ScheduledPriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
            TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, 
            new PriorityBlockingQueue<Runnable>(INITIAL_QUEUE_SIZE, new PriorityTaskComparator()), threadFactory);
    }

    public ScheduledPriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
            TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, 
            new PriorityBlockingQueue<Runnable>(INITIAL_QUEUE_SIZE, new PriorityTaskComparator()), threadFactory, handler);
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
        // TODO Auto-generated method stub
        return null;
    }

}

正如您所看到的，构造函数问题已经解决，但仍然留下了调度方法的实现ScheduledExecutorService。

所以我问你，有没有什么方法可以将 a 传递Comparator到队列，或者有一种简单但不太详尽的方法来创建一个自己的执行器类，该执行器类实现类中的方法ScheduledExecutorService并提供类中的方法ThreadPoolExecutor以及使用 a PriorityBlockingQueue？

Answer 1

我寻找其他可能的解决方案，得到以下结果：

由于 aThreadPoolExecutor管理多个线程池Threads（即，如果您在Executors.newFixedThreadPool(int nThreads)方法中设置两个或多个线程），并且如果您确实想混合基于BlockingQueue它的优先级，那么我建议执行以下操作：

• 创建ThreadPoolExecutor一个与上面类似的自己的类，使用PriorityBlockingQueue。
• 创建您自己的Task类（或FutureTask扩展，无论您认为最适合您）
• 这些任务类适用于一次性任务，这意味着它们只运行一次。

对于应该定期在后台运行的循环任务，我为此想出了一个简单的类：

public abstract class AbstractThread extends Thread {

    protected Runnable runnable;

    protected AbstractThread(String name, Runnable runnable) {
        super(runnable, name);

        this.runnable = runnable;
    }

    /**
     * This method provides a way to perform some action before the thread is actually starting.
     */
    protected abstract void beforeExecution();

    /**
     * This method provides a way to perform some action after the thread finished.
     */
    protected abstract void afterExecution();

    @Override
    public void run() {
        try {
            doRun();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Run the given runnable here.
     * 
     * @throws InterruptedException 
     */
    protected abstract void doRun() throws InterruptedException;

}

虽然简单的一次性线程只运行可运行一次：

@Override
protected void doRun() {
    beforeExecution();

    runnable.run();

    afterExecution();
}

线程中的周期性任务只会执行以下操作：

@Override
protected void doRun() throws InterruptedException {
    beforeExecution();

    while(!isInterrupted()) {
        runnable.run();
        Thread.sleep(millis);
    }

    afterExecution();
}

如果您想支持偶尔运行一次的周期性任务，您可以将延迟参数传递给实例，或者只在可运行对象中Thread编写类似的内容。Thread.sleep(delay)

这不是实际的代码，只是一个建议，因为我现在正在尝试使用它。