死信队列和延迟队列_在实践中使用延迟队列

死信队列和延迟队列

通常，在某些情况下，当您有某种工作或作业队列时，有必要不立即处理每个工作项或作业，而是要延迟一些时间。例如，如果用户单击一个按钮来触发要完成的某项工作，而一秒钟后，用户意识到他/她错了，则该工作根本就不会开始。或者，如果有一个用例，则在延迟（过期）后应删除队列中的某些工作元素。

有很多实现，但是我想描述的是使用纯JDK并发框架类： DelayedQueue和Delayed接口。

让我从定义工作项的简单（空）界面开始。我跳过诸如属性和方法之类的实现细节，因为它们并不重要。

package com.example.delayed;public interface WorkItem {// Some properties and methods here
}

我们模型中的下一个类将代表被推迟的工作项并实现Delayed接口。仅需考虑几个基本概念：延迟本身和相应工作项已提交的实际时间。这就是到期日的计算方式。因此，我们通过引入PostponedWorkItem类来做到这一点。

package com.example.delayed;import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;public class PostponedWorkItem implements Delayed {private final long origin;private final long delay;private final WorkItem workItem;public PostponedWorkItem( final WorkItem workItem, final long delay ) {this.origin = System.currentTimeMillis();this.workItem = workItem;this.delay = delay;}@Overridepublic long getDelay( TimeUnit unit ) {return unit.convert( delay - ( System.currentTimeMillis() - origin ), TimeUnit.MILLISECONDS );}@Overridepublic int compareTo( Delayed delayed ) {if( delayed == this ) {return 0;}if( delayed instanceof PostponedWorkItem ) {long diff = delay - ( ( PostponedWorkItem )delayed ).delay;return ( ( diff == 0 ) ? 0 : ( ( diff < 0 ) ? -1 : 1 ) );}long d = ( getDelay( TimeUnit.MILLISECONDS ) - delayed.getDelay( TimeUnit.MILLISECONDS ) );return ( ( d == 0 ) ? 0 : ( ( d < 0 ) ? -1 : 1 ) );}
}

如您所见，我们创建该类的新实例，并将当前系统时间保存在内部origin属性中。 getDelayed方法计算工作项过期之前剩余的实际时间。延迟是外部设置，它是构造函数参数。由于Delayed扩展了此接口，因此必须强制实施Comparable <Delayed> 。

现在，我们大部分完成了！为了完成该示例，让我们通过实现equals和hashCode来确保不会将相同的工作项两次提交到工作队列中（实现非常简单，不需要任何注释）。

public class PostponedWorkItem implements Delayed {...@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + ( ( workItem == null ) ? 0 : workItem.hashCode() );return result;}@Overridepublic boolean equals( Object obj ) {if( this == obj ) {return true;}if( obj == null ) {return false;}if( !( obj instanceof PostponedWorkItem ) ) {return false;}final PostponedWorkItem other = ( PostponedWorkItem )obj;if( workItem == null ) {if( other.workItem != null ) {return false;}} else if( !workItem.equals( other.workItem ) ) {return false;}return true;}
}

最后一步是引入某种管理器，该管理器将安排工作项并定期轮询过期的项：见WorkItemScheduler类。

package com.example.delayed;import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;public class WorkItemScheduler {private final long delay = 2000; // 2 secondsprivate final BlockingQueue< PostponedWorkItem > delayed =new DelayQueue< PostponedWorkItem >(); public void addWorkItem( final WorkItem workItem ) {final PostponedWorkItem postponed = new PostponedWorkItem( workItem, delay );if( !delayed.contains( postponed )) {delayed.offer( postponed );}}public void process() {final Collection< PostponedWorkItem > expired = new ArrayList< PostponedWorkItem >();delayed.drainTo( expired );for( final PostponedWorkItem postponed: expired ) {// Do some real work here with postponed.getWorkItem()}}
}

使用BlockingQueue可以确保线程安全和高级别的并发性。该处理方法应定期运行，以排干工作项目队列。它可以通过Spring Framework中的@ Scheduled注释或JEE 6中 EJB的@Schedule注释进行注释。

请享用！

参考：在Andriy Redko {devmind}博客上，我们的JCG合作伙伴 Andriy Redko 在实践中使用了延迟队列。