为什么我无法构建一个由DelayQueue支持的ThreadPoolExecutor？

Question

我正在尝试创建一个ThreadPoolExecutor：

// Thingy implements Delayed and Runnable
ExecutorService executor = new ThreadPoolExecutor(1, 1, 0l, TimeUnit.SECONDS, new DelayQueue<Thingy>());

编译器提示“找不到符号”：

symbol  : constructor ThreadPoolExecutor(int,int,long,java.util.concurrent.TimeUnit,java.util.concurrent.DelayQueue<Thingy>)

但我不明白—— DelayQueue 实现了 BlockingQueue，所以我应该能够使用 这个构造函数 吧？

- Avi Flax

Answer 1

这是一个泛型问题。您不能使用DelayQueue<Thingy>，而必须使用DelayQueue<Runnable>，因为ThreadPoolExecutor构造函数未声明接受Runnable子类型的队列。

- Dan Dyer

---

谢谢！哥们儿，泛型太疯狂了。 - Avi Flax

我仍然需要帮助。我将其更改为new DelayQueue<Runnable>()，但这也不起作用，因为DelayQueue被声明为DelayQueue <E extends Delayed>，所以显然我需要指定一些扩展Delayed的内容，但我*还需要指定Runnable - 这太疯狂了。请帮忙！ - Avi Flax

@Avi 说得好，我没有注意到这一点。在这种情况下，我认为你要么放弃 DelayQueue，要么使用未经检查的强制类型转换。如果 ThreadPoolExecutor 被声明接受 BlockingQueue<? extends Runnable> ，那就没问题了。 - Dan Dyer

Answer 2

虽然这是一个老问题，但我想发表我的答案，因为最近我正在寻找这个解决方案。可以在ThreadPoolExecutor后面使用DelayQueue，只需要一点代码包装即可。关键是让DelayQueue作为BlockingQueue呈现出来。

我首先定义了一个接口DR，它扩展了Runnable和Delayed。请注意，这里的静态方法创建DR的实例（未显示Instance类）。

public interface DR extends Delayed, Runnable {    
    public static DR make( Runnable r )
    {
        if (r instanceof DR)
        {
            return (DR)r;
        }
        Impl impl = new Impl(r);
        if (r instanceof Delayed)
        {
            impl.expires = ((Delayed) r).getDelay( TimeUnit.MILLISECONDS );
        }
        return impl;
    }

    public static DR make( Runnable r, long expires )
    {
        if (r instanceof DR)
        {
            if (expires == ((DR)r).getDelay( TimeUnit.MILLISECONDS ))
            {
                return (DR)r;
            }
        }
        return new Impl(r, expires);
    }
}

实现应该覆盖以下方法：public int compareTo(Delayed o)，public boolean equals(Object o)和public int hashCode()。

创建一个扩展DelayQueue的类。此类添加了一个单一的方法，将DelayQueue呈现为BlockingQueue。返回的类简单地包装了DelayQueue，并使用DR接口的make方法在必要时从Runnable转换为DR。

public class DelayedBlockingQueue extends DelayQueue<DR>  {

    public BlockingQueue<Runnable> asRunnableQueue() {
        return new BlockingQueue<Runnable>(){
            DelayedBlockingQueue dbq = DelayedBlockingQueue.this;
            public boolean add(Runnable e) {
                return dbq.add( DR.make( e ));
            }

            private List<DR> makeList( Collection<? extends Runnable> coll)
            {
               return coll.stream().map( r -> DR.make( r ) ).collect( Collectors.toList() ) ;
            }

            public boolean addAll(Collection<? extends Runnable> arg0) {
                return dbq.addAll(makeList( arg0 ) );
            }

            public void clear() {
                dbq.clear();
            }

            public boolean contains(Object o) {
                if (o instanceof Runnable) { 
                    return dbq.contains( DR.make( (Runnable)o ) );
                } 
                return false;
            }

            public boolean containsAll(Collection<?> arg0) {
                List<DR> lst = new ArrayList<DR>();
                for (Object o : arg0)
                {
                    if (o instanceof Runnable)
                    {
                        lst.add(  DR.make(  (Runnable)o ) );
                    }
                    else {
                        return false;
                    }
                }

                return dbq.containsAll( lst );
            }

            public int drainTo(Collection<? super Runnable> c, int maxElements) {
                return dbq.drainTo( c, maxElements );
            }

            public int drainTo(Collection<? super Runnable> c) {
                return dbq.drainTo( c );
            }

            public Runnable element() {
                return dbq.element();
            }                

            public void forEach(Consumer<? super Runnable> arg0) {
                dbq.forEach( arg0 );
            }

            public boolean isEmpty() {
                return dbq.isEmpty();
            }

            public Iterator<Runnable> iterator() {
                return WrappedIterator.create( dbq.iterator() ).mapWith( dr -> (Runnable)dr );
            }

            public boolean offer(Runnable e, long timeout, TimeUnit unit) throws InterruptedException {
                return dbq.offer( DR.make( e ), timeout, unit );
            }

            public boolean offer(Runnable e) {
                return dbq.offer( DR.make( e ) );
            }

            public Runnable peek() {
                return dbq.peek();
            }

            public Runnable poll() {
                return dbq.poll();
            }

            public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
                return dbq.poll( timeout, unit );
            }

            public void put(Runnable e) throws InterruptedException {
                dbq.put( DR.make(e) );
            }

            public int remainingCapacity() {
                return dbq.remainingCapacity();
            }

            public Runnable remove() {
                return dbq.remove();
            }

            public boolean remove(Object o) {
                if (o instanceof Runnable)
                {
                    return dbq.remove( DR.make(  (Runnable)o) );
                }
                return false;
            }

            public boolean removeAll(Collection<?> arg0) {
                List<DR> lst = new ArrayList<DR>();
                for (Object o : arg0)
                {
                    if (o instanceof Runnable)
                    {
                        lst.add(  DR.make(  (Runnable)o ) );
                    }

                }               
                return dbq.removeAll( lst );
            }

            public boolean retainAll(Collection<?> arg0) {
                return dbq.retainAll( arg0 );
            }

            public int size() {
                return dbq.size();
            }

            public Runnable take() throws InterruptedException {
                return dbq.take();
            }

            public Object[] toArray() {
                return dbq.toArray();
            }

            public <T> T[] toArray(T[] arg0) {
                return dbq.toArray( arg0 );
            }                
    };                       
}

要使用该解决方案，请创建DelayedBlockingQueue，并使用asRunnableQueue()方法将可运行队列传递给ThreadPoolExecutor构造函数。

DelayedBlockingQueue queue = new DelayedBlockingQueue();
ThreadPoolExecutor execService = new ThreadPoolExecutor( 1, 5, 30, TimeUnit.SECONDS, queue.asRunnableQueue() );

- cWarren

---

请注意，我尝试了这个解决方案，但它并不起作用。如果您的线程没有忙碌，ThreadPoolExecutor 将完全绕过队列。 - Robert

Answer 3

RunnableScheduledFuture 是可运行和延迟的，但它不能转换为 BlockingQueue<Runnable>。请参见 Java教程 了解原因。

看一下 ScheduledThreadPoolExecutor，它可以安排命令在给定延迟后运行，或定期执行。

- Pierre