具有优先级任务的线程池执行器，并避免饥饿

Question

对于我的用例，我需要一个可以根据优先级执行任务的executor。实现这一点的简单方法是使用具有PriorityBlockingQueue的线程池并覆盖newTaskFor()，以返回基于任务优先级可比较的自定义未来任务。

//Define priorities
public enum Priority {
    HIGH, MEDIUM, LOW, VERYLOW;
}

优先任务

//A Callable tasks that has priority. Concrete implementation will implement 
//call() to do actual work and getPriority() to return priority
public abstract class PriorityTask<V> implements Callable<V> {
    public abstract Priority getPriority ();
}

实际的执行器实现

public class PriorityTaskThreadPoolExecutor <V> {
    int _poolSize;
    private PriorityBlockingQueue<Runnable> _poolQueue = 
                                       new PriorityBlockingQueue<Runnable>(500); 
    private ThreadPoolExecutor _pool;

    public PriorityTaskThreadPoolExecutor (int poolSize) {
        _poolSize = poolSize;

        _pool = new ThreadPoolExecutor(_poolSize, _poolSize, 5, TimeUnit.MINUTES, 
                                      _poolQueue) {
                        //Override newTaskFor() to return wrap PriorityTask 
                        //with a PriorityFutureTaskWrapper.
                        @Override
                        protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {
                            return new PriorityFutureTaskWrapper<V>((PriorityTask<V>) c);
                        }
                };

        _pool.allowCoreThreadTimeOut(true);
    }

    public Future<V> submit (PriorityTask<V> task) {
        return _pool.submit(task);
    }

}

//A future task that wraps around the priority task to be used in the queue
class PriorityFutureTaskWrapper<V> extends FutureTask<V> 
                             implements Comparable <PriorityFutureTaskWrapper<V>> {
    PriorityTask<V> _priorityTask;

    public PriorityFutureTaskWrapper (PriorityTask<V> priorityTask) {
        super(priorityTask);
        _priorityTask = priorityTask;
    }

    public PriorityTask<V> getPriorityTask () {
        return _priorityTask;
    }

    @Override
    public int compareTo(PriorityFutureTaskWrapper<V> o) {
        return _priorityTask.getPriority().ordinal() - 
               o.getPriorityTask().getPriority().ordinal();
    }
}

这样做的问题是，在我的用例中，低优先级任务有可能永远处于饥饿状态。我想避免这种情况。我找不到一种干净的方法来使用java中可用的执行器/池来做到这一点。因此，我正在考虑编写自己的executor。我有两种不同的方法。

1)带PriorityBlockingQueue的自定义线程池。将有一个单独的线程，用于检查队列中的任务期限。较旧的任务将被删除，并以提升的优先级重新添加。

2)我的用例将只有有限数量的优先级，比如1-4。对于每个优先级，我将有4个不同的队列。现在，当定制池中的线程必须处理下一个任务时，它将按以下顺序扫描队列，而不是阻塞队列。

40%的线程- Q1、Q2、Q3、Q4

30%的线程- Q2、Q1、Q3、Q4

20%的线程- Q3、Q1、Q2、Q4

10%的线程- Q4、Q1、Q2、Q3

当线程被通知队列中有新的添加时，或者当该线程执行的当前任务完成时，它将完成扫描。其他时候，线程将会等待。但是，与队列上的阻塞相比，扫描的效率会稍低一些。

Apprach 2更适合我的用例。

有没有人尝试过这些方法中的任何一种，或者针对类似的用例尝试过不同的方法？有什么想法/建议吗？

Answer 1

here已经讨论过，没有简单的方法来更改PriorityQueue中已插入元素的优先级

第二种选择应该易于实现，比如处理高优先级队列中的任务比处理低优先级队列中的任务多

您还可以考虑为每个优先级设置不同的ThreadPools，每个池中的线程数量取决于任务的优先级。