Executor 和 PriorityBlockingQueue 实现问题上的 AsyncTask

Question

在this SO question 的浪潮中，以及来自another one 的许多提示，我正在尝试实现一个AsyncTask 变体，其中包含可以优先处理的任务。

在我的CustomAsyncTask 课程中，我有：

public abstract class CustomAsyncTask<Params, Progress, Result> {

    private static int CORE_POOL_SIZE = 1;
    private static int MAXIMUM_POOL_SIZE = 1;

    private static final int KEEP_ALIVE = 1;

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

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

    private static final BlockingQueue<DownloadTask> pPoolWorkQueue =
            new PriorityBlockingQueue<DownloadTask>(10, new DownloadTasksComparator());

    @SuppressWarnings({ "unchecked", "rawtypes" })
    public static Executor PRIORITY_THREAD_POOL_EXECUTOR
            = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE,
                    TimeUnit.SECONDS, (PriorityBlockingQueue) pPoolWorkQueue, sThreadFactory);

   //...
}

比较器：

public class DownloadTasksComparator implements Comparator<DownloadTask> {

    @Override
    public int compare(DownloadTask arg0, DownloadTask arg1) {
        int res;

        if (arg0 == null && arg1 == null) {
            res = 0;
        } else if (arg0 == null) {
            res = -1;
        } else if (arg1 == null) {
            res = 1;
        }

        res = arg0.getPriority() - arg1.getPriority();

        return res;
    }
}

在扩展CustomAsyncTask 的DownloadTask 类中，我有一个priority 整数字段和一个getPriority() 方法。

我将任务执行称为：

DownloadTask dt = new DownloadTask(..., PRIORITY_NORMAL, ...);
dt.executeOnExecutor(CustomAsyncTask.PRIORITY_THREAD_POOL_EXECUTOR);

这是可行的：如果池大小为 1，则下载将一一执行；如果池大小为 2，等等。

注意：priority 整数具有任意值：

public static final int PRIORITY_HIGH = 10;
public static final int PRIORITY_NORMAL = 1;

但如果我将任务称为：

DownloadTask dt = new DownloadTask(..., PRIORITY_HIGH, ...);
dt.executeOnExecutor(CustomAsyncTask.PRIORITY_THREAD_POOL_EXECUTOR);

我有一个java.lang.ClassCastException: my.pkg.name.CustomAsyncTask$3 cannot be cast to my.pkg.name.DownloadTask

然后

at my.pkg.name.DownloadTasksComparator.compare(DownloadTasksComparator.java:1)
at java.util.concurrent.PriorityBlockingQueue.siftUpUsingComparator(PriorityBlockingQueue.java:334)
at java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:447)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1295)
at my.pkg.name.CustomAsyncTask.executeOnExecutor(CustomAsyncTask.java:494)
at my.pkg.name.GetDownloadTaskListener$1.finishDownload(GetDownloadTaskListener.java:180)
at my.pkg.name.DownloadTask.onPostExecute(DownloadTask.java:330)
at my.pkg.name.DownloadTask.onPostExecute(DownloadTask.java:1)
at my.pkg.name.CustomAsyncTask.finish(CustomAsyncTask.java:536)
at my.pkg.name.CustomAsyncTask.access$0(CustomAsyncTask.java:532)
at my.pkg.name.CustomAsyncTask$InternalHandler.handleMessage(CustomAsyncTask.java:549)
at android.os.Handler.dispatchMessage(Handler.java:99)
at android.os.Looper.loop(Looper.java:137)
at android.app.ActivityThread.main(ActivityThread.java:4745)
at java.lang.reflect.Method.invokeNative(Native Method)
at java.lang.reflect.Method.invoke(Method.java:511)
at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:786)
at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:553)
at dalvik.system.NativeStart.main(Native Method)

全部来自AndroidRuntime

我真的不知道...

编辑：此时，我已经封装了一个小型 Eclipse 项目，该项目以与大型应用程序完全相同的方式实现事物并遇到相同的问题。它逐字借用CustomAsyncTaskComparator 和CustomAsyncTask。没有给出视觉反馈。应用程序的进度只是一些 LogCat 输出。但它给出了这个想法。当超过两个任务入队时，应用 FCs。

https://www.dropbox.com/s/lrg4kscgw3f1xwr/ConcurrentTest.tar.gz

Answer 1

CustomAsyncTask 和AsyncTask 一样，使用static 队列。 目前，该队列中的所有内容都将通过您的DownloadTasksComparator 运行。但是，DownloadTasksComparator 仅适用于 DownloadTask。只要你只使用DownloadTask，而不是CustomAsyncTask 的其他子类，你就可以了。但是，显然您还有其他一些匿名内部类扩展 CustomAsyncTask，而这不是 DownloadTask。

将CustomAsyncTask 设为abstract，将getPriority() 方法作为abstract 方法。将 DownloadTasksComparator 重命名为 CustomAsyncTaskComparator 并让它比较 CustomAsyncTask 实例。然后，CustomAsyncTask 的其他子类需要实现它们自己的 getPriority() 方法，以使它们能够与工作队列中的 DownloadTask 实例一起排序。

Answer 2

您在查看AsyncTask 实现时可能已经注意到，它在内部使用FutureTask 来处理后台任务，这就是交给Executor 并可能在其工作队列中排队的内容。由于您已经在派生自己的实现，因此您可以将 FutureTask 替换为包含对 AsyncTask 的引用的自定义派生，以便从您的 Comparator 实现中访问。

此外，不要替换自定义 AsyncTask 衍生的默认静态 Executor，而应在子类中使用 executeOnExecutor() 方法，以便可以以通用方式使用它。

Answer 3

问题是你用来构造 ThreadPoolExecutor 的PriorityBlockingQueue<CustomAsyncTask>。 ThreadPoolExecutor 应该接受BlockingQueue<Runnable>。 ThreadPoolExecutor 使用此队列对可运行对象进行排队（可运行对象通过调用execute(Runnable r) 提交）。

在 AsyncTask 的情况下，它使用 FutureTask 对象调用Executor.execute(Runnable r)，因此执行器排队的是这个 FutureTask 对象，而不是 AsyncTask 本身。因此，当比较器尝试将 runnable 强制转换为 DownloadTask 时，就会抛出异常。我猜my.pkg.name.CustomAsyncTask$3 可能是一个内部可运行类。

这里是AsyncTask的部分源码：

public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
        Params... params) {
    if (mStatus != Status.PENDING) {
        switch (mStatus) {
            case RUNNING:
                throw new IllegalStateException("Cannot execute task:"
                        + " the task is already running.");
            case FINISHED:
                throw new IllegalStateException("Cannot execute task:"
                        + " the task has already been executed "
                        + "(a task can be executed only once)");
        }
    }

    mStatus = Status.RUNNING;

    onPreExecute();

    mWorker.mParams = params;
    exec.execute(mFuture);

    return this;
}

如果您尝试使用此 PriorityBlockingQueue 方法以优先级运行 AsyncTask，则应将传递给 execute 方法的可运行对象子类化，使其具有优先级。您正在使用自定义异步任务，因此您可以完全控制变量。

对于那些试图优先运行原始 AsyncTask 的人，我认为没有什么可以做的。 AsyncTask“执行”一个私有成员变量mFuture，你的双手被束缚了。

我目前正在处理类似的问题，但在我的情况下，“优先级”是时间戳，我可以在可运行的实现中设置它，因此它不受 AsyncTask 的限制。

这是我的代码，以防万一它可能对某人有所帮助。对不起我的英语不好！

/* Executor used by AsyncTask. Use it with executeOnExecutor()*/
class PriorityExecutor extends ThreadPoolExecutor{

    //workQueue is a instance of PriorityBlockingQueue<Runnable>
    public PriorityExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory);

    }

    //wrap the runnable passed in.
    @Override
    public void execute(Runnable command) {
        super.execute(new PriorityRunnableWrapper(command));
    }
}

//a wrapper class
class PriorityRunnableWrapper implements Runnable ,Comparable<PriorityRunnableWrapper>{

    long addedTime;//the "priority"
    Runnable r;

    public PriorityRunnableWrapper(Runnable r){
        this.r = r;
        addedTime = System.nanoTime();//in my case the timestamp is the priority
    }

    @Override
    public void run() {
        r.run();
    }

    @Override
    public int compareTo(PriorityRunnableWrapper another) {
        if(addedTime == another.addedTime)return 0;
        return addedTime - another.addedTime > 0 ? -1 : 1;
    }

}