如何使用ThreadPoolExecutor和自定义任务实现PriorityBlockingQueue [英] How to implement PriorityBlockingQueue with ThreadPoolExecutor and custom tasks

问题描述

我经常搜索，但找不到解决问题的方法。

I've searched a lot but could not find a solutuion to my problem.

我有自己的类 BaseTask ，它使用 ThreadPoolExecutor b来处理任务。

如果我不想确定优先级（即使用 LinkedBlockingQueue ），这个工作正常，但当我尝试使用 PriorityBlockingQueue 我得到 ClassCastException 因为 ThreadPoolExecutor 包裹了我的任务到 FutureTask 对象。

这显然没问题，因为 FutureTask 没有实现可比较，但我如何继续解决优先级问题？

我读过你可以覆盖 newTaskFor in ThreadPoolExecutor ，但我似乎无法找到这个方法......？

I have my own class, BaseTask, that uses a ThreadPoolExecutor to handle tasks.
If I don't want prioritization (i.e. using a LinkedBlockingQueue) this works just fine, but when I try to use a PriorityBlockingQueue I get ClassCastException because the ThreadPoolExecutor wraps my Tasks into a FutureTask object.
This is obviously OK because the FutureTaskdoes not implement Comparable, but how would I go on to solve the priority problem?
I've read that you could override newTaskFor in ThreadPoolExecutor, but I can not seem to find this method at all...?

非常感谢任何建议！

一些代码可以提供帮助：

在我的 BaseTask 类中，我有

private static final BlockingQueue<Runnable> sWorkQueue = new PriorityBlockingQueue<Runnable>();

private static final ThreadFactory sThreadFactory = new ThreadFactory() {
    private final AtomicInteger mCount = new AtomicInteger(1);

    public Thread newThread(Runnable r) {
        return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
    }
};

private static final BaseThreadPoolExecutor sExecutor = new BaseThreadPoolExecutor(
    1, Integer.MAX_VALUE, 10, TimeUnit.SECONDS, sWorkQueue, sThreadFactory);

private final BaseFutureTask<Result> mFuture;

public BaseTask(int priority) {
    mFuture = new BaseFutureTask<Result>(mWorker, priority);
}

public final BaseTask<Params, Progress, Result> execute(Params... params) {

    /* Some unimportant code here */

    sExecutor.execute(mFuture);
}

在 BaseFutureTask 类

@Override
public int compareTo(BaseFutureTask another) {
    long diff = this.priority - another.priority;

    return Long.signum(diff);
}

在 BaseThreadPoolExecutor class我重写了3 submit 方法...

此类中的构造函数被调用，但都没有提交方法

In BaseThreadPoolExecutor class i override the 3 submit methods...
The constructor in this class gets called, but none of the submit methods

推荐答案

public class ExecutorPriority {

public static void main(String[] args) {

    PriorityBlockingQueue<Runnable> pq = new PriorityBlockingQueue<Runnable>(20, new ComparePriority());

    Executor exe = new ThreadPoolExecutor(1, 2, 10, TimeUnit.SECONDS, pq);
    exe.execute(new RunWithPriority(2) {

        @Override
        public void run() {

            System.out.println(this.getPriority() + " started");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ExecutorPriority.class.getName()).log(Level.SEVERE, null, ex);
            }
            System.out.println(this.getPriority() + " finished");
        }
    });
    exe.execute(new RunWithPriority(10) {

        @Override
        public void run() {
            System.out.println(this.getPriority() + " started");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ExecutorPriority.class.getName()).log(Level.SEVERE, null, ex);
            }
            System.out.println(this.getPriority() + " finished");
        }
    });

}

private static class ComparePriority<T extends RunWithPriority> implements Comparator<T> {

    @Override
    public int compare(T o1, T o2) {
        return o1.getPriority().compareTo(o2.getPriority());
    }
}

}

你可以猜到RunWithPriority是一个抽象类，它是Runnable并且有一个Integer优先级字段

as you can guess RunWithPriority is an abstract class that is Runnable and has a Integer priority field

这篇关于如何使用ThreadPoolExecutor和自定义任务实现PriorityBlockingQueue的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！