文档章节

ThreadPoolExecutor的PriorityBlockingQueue支持问题

canghailan
 canghailan
发布于 2011/12/12 17:14
字数 926
阅读 2747
收藏 4

最近在使用ThreadPoolExecutor时遇到一个问题:当ThreadPoolExecutor使用的BlockingQueue为PriorityBlockingQueue时,会出现异常,原因是java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable。Google之,发现有很多同样的问题,但没有给出解决方案,只能查看源代码以期能找到并解决问题。

首先根据Exception找到问题原因:

Exception in thread "main" java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable
	at java.util.concurrent.PriorityBlockingQueue.siftUpComparable(PriorityBlockingQueue.java:347)
	at java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:475)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1329)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:132)
...

找到java.util.concurrent.PriorityBlockingQueue.siftUpComparable方法:

private static <T> void siftUpComparable(int k, T x, Object[] array) {
        Comparable<? super T> key = (Comparable<? super T>) x;
        while (k > 0) {
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (key.compareTo((T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

是在Comparable<? super T> key = (Comparable<? super T>) x;上出现问题,根据java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable知道x的类型是java.util.concurrent.FutureTask。现在看看FutureTask:

public class FutureTask<V> implements RunnableFuture<V> {
...

public interface RunnableFuture<V> extends Runnable, Future<V> {
...

public
interface Runnable {
...

public interface Future<V> {
...

可见FutureTask的确没有实现Comparable接口。但是我所提交的Task

public static class Task implements Callable<Integer>, Comparable<Task> {
...

是实现了Comparable接口的。我的Task为什么变成了FutureTask?只好找到ThreadPoolExecutor的submit(Callable<T> task)方法一看究竟,它是在AbstractExecutorService中实现的。

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

可以看到通过newTaskFor方法,我所提交的Task变成了FutureTask:

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

OK,看来问题就是出在FutureTask这个类上了:

public class FutureTask<V> implements RunnableFuture<V> {
    private final Sync sync;

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
    }
...
    private final class Sync extends AbstractQueuedSynchronizer {
        private final Callable<V> callable;

        Sync(Callable<V> callable) {
            this.callable = callable;
        }
...

看来他是将我提交的Task的Comparable接口直接忽略了。现在的问题就变成了让FutureTask支持Comparable接口,最简单的方法是用一个ComparableFutureTask继承FutureTask并实现Comparable接口,但也必须要Override ThreadPoolExecutor的newTaskFor方法,显得有些麻烦。为此又继续Google,却没有发现一些更好的方法。先实现了再说。

package canghailan.util.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
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;

/**
 * @author canghailan
 * @datetime 2011-12-10 13:57:19
 */
public class XThreadPoolExecutor extends ThreadPoolExecutor {

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

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

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new ComparableFutureTask<>(runnable, value);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new ComparableFutureTask<>(callable);
    }

    protected class ComparableFutureTask<V>
            extends FutureTask<V> implements Comparable<ComparableFutureTask<V>> {
        private Object object;

        public ComparableFutureTask(Callable<V> callable) {
            super(callable);
            object = callable;
        }

        public ComparableFutureTask(Runnable runnable, V result) {
            super(runnable, result);
            object = runnable;
        }

        @Override
        @SuppressWarnings("unchecked")
        public int compareTo(ComparableFutureTask<V> o) {
            if (this == o) {
                return 0;
            }
            if (o == null) {
                return -1; // high priority
            }
            if (object != null && o.object != null) {
                if (object.getClass().equals(o.object.getClass())) {
                    if (object instanceof Comparable) {
                        return ((Comparable) object).compareTo(o.object);
                    }
                }
            }
            return 0;
        }
    }
}

测试一下:

package canghailan;

import canghailan.util.concurrent.XThreadPoolExecutor;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author canghailan
 * @datetime 2011-12-10 7:09:39
 */
public class TestXThreadPoolExecutor {

    public static void main(String args[]) throws Exception {
        Phaser phaser = new Phaser(1);
        ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
        ExecutorService service = new XThreadPoolExecutor(
                5, 5, 0L, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>());
        for (int i = 0; i < 10; ++i) {
            phaser.register();
            service.submit(new Task(i, queue, phaser));
        }
        phaser.arriveAndAwaitAdvance();
        System.out.println(queue);
        service.shutdown();
    }

    public static class Task implements Callable<Integer>, Comparable<Task> {
        private Integer id;
        private ConcurrentLinkedQueue<Integer> queue;
        private Phaser phaser;

        public Task(Integer id, ConcurrentLinkedQueue<Integer> queue, Phaser phaser) {
            this.id = id;
            this.queue = queue;
            this.phaser = phaser;
        }

        @Override
        public Integer call() throws Exception {
            queue.offer(id);
            phaser.arrive();
            return id;
        }

        @Override
        public int compareTo(Task o) {
            return -id.compareTo(o.id);
        }
    }
}

输出为[0, 1, 2, 3, 9, 8, 7, 6, 5, 4],[0, 1, 2, 3, 4, 7, 6, 5, 9, 8],[0, 1, 2, 3, 4, 7, 6, 5, 9, 8]...。

根据结果来看,的确起作用了。先到此为止,如果有更好的办法,请不吝赐教。

© 著作权归作者所有

canghailan

canghailan

粉丝 28
博文 14
码字总数 10765
作品 0
武汉
程序员
私信 提问
加载中

评论(1)

千里山南
千里山南
遇到同样问题,感谢分享
ThreadPoolExecutor的配置(二)

配置ThreadPoolExecutor二 线程的创建和销毁 线程池的基本大小(Core Pool Size)、最大大小(Maximum Pool Size)以及存活时间(keepAliveTime)等因素共同负责线程的创建和销毁。基本大小也...

秋风醉了
2016/09/19
0
0
java.until.concurrent(一)

一 线程池 2.1 线程池相关类结构 Executor ExecutorService AbstractExecutorService ThreadPoolExecutor 2.2 各个类介绍 2.2.1 接口Executor 接口Executor 是一个简单的标准化接口,用于定义...

莫问viva
2015/06/19
0
0
干货,谈谈对java线程池的理解

线程池的概念 线程池由任务队列和工作线程组成,它可以重用线程来避免线程创建的开销,在任务过多时通过排队避免创建过多线程来减少系统资源消耗和竞争,确保任务有序完成;ThreadPoolExecu...

Code_Steven
2018/08/08
0
0
Java进阶——— 线程池的原理分析

前言 在了解线程池之前,其实首先出现的疑问是:为什么要使用线程池,其次是了解什么是线程池,最后是如何使用线程池,带着疑问去学习。 为什么要使用 前面多线程文章中,需要使用线程就开启...

T9's
01/22
0
0
Android性能优化之使用线程池处理异步任务

说到线程,我想大家都不陌生,因为在开发时候或多或少都会用到线程,而通常创建线程有两种方式: 1、继承Thread类 2、实现Runnable接口 虽说这两种方式都可以创建出一个线程,不过它们之间还...

u010687392
2015/11/17
0
0

没有更多内容

加载失败,请刷新页面

加载更多

windows 远程下载sftp 服务器 文件

下载 putty https://www.chiark.greenend.org.uk/~sgtatham/putty/latest.html 把安装后的 psftp.exe 放到需要下载文件的目录 尝试执行远程连接 运行 psftp.exe open : 111:111:111:111 录入...

donald121
16分钟前
2
0
阿里、网易和腾讯面试题 C/C++

一、线程、锁 1、Posix Thread互斥锁 线程锁创建 a.静态创建 pthread_mutex_tmutex = PTHREAD_MUTEX_INITIALIZER; b.动态创建 pthread_mutex_tmutex = pthread_mutex_init(pthread_mutex_t*......

linux服务器架构
17分钟前
3
0
Linux搭建spark单机操作环境

https://blog.csdn.net/king13127/article/details/86528416https://www.cnblogs.com/duanxz/p/4437802.html##spark、Hadoop、java、scalahttps://blog.csdn.net/pucao_cug/article/det......

KYO4321
20分钟前
1
0
C语言动态内存管理和动态内存分配

动态内存管理同时还具有一个优点:当程序在具有更多内存的系统上需要处理更多数据时,不需要重写程序。标准库提供以下四个函数用于动态内存管理: (1) malloc()、calloc() 分配新的内存区...

阿里云云栖社区
24分钟前
7
0
浅谈Java队列Queue

队列Queue就是一个先进先出的数据结构,与List、Set同一级别,继承了Collection接口。 一、Queue的实现 1、阻塞队列(BlockingQueue) ① 插入:队列不满时可执行插入元素线程,直到队列满。...

青衣霓裳
41分钟前
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部