java.util.concurrent.DelayQueue忽略过期元素

Question

下面的java代码示例使用java DelayQueue来处理任务。然而，从另一个线程插入任务似乎会破坏（我）预期的行为。

道歉，该代码例如是如此长，但在总结：

1. 主线程添加5个任务（AE）与各种延迟一个DelayQueue（0毫秒，10毫秒，100毫秒1000毫秒，10000ms）
2. 另一胎开始它增加了后3000ms
3. 另一个任务的DelayQueue主线程轮询DelayQueue并在每个任务报告到期
4. 后8000MS主线程报告中剩余的DelayQueue
任务

，我从代码示例得到的输出是：

------initial tasks --------------- 
task A due in 0ms 
task B due in 9ms 
task C due in 99ms 
task D due in 999ms 
task E due in 9999ms 
task F due in 99999ms 
------processing-------------------- 
time = 5 task A due in -1ms 
time = 14 task B due in 0ms 
time = 104 task C due in 0ms 
time = 1004 task D due in 0ms 
time = 3003 added task Z due in 0ms 
------remaining after 15007ms ----------- 
task F due in 84996ms 
task E due in -5003ms 
task Z due in -12004ms 

我的问题是：为什么后15000ms都存在过期留在DelayQueue任务（即其中GetDelay（）返回一个值-ve） ？

，我查了一些事情：

• 我已经实现的compareTo（）来定义任务的自然顺序
• equals（）方法是用的compareTo（）
• hashCode（）方法是一贯的重写

我会对如何解决这个问题感兴趣。预先感谢您的帮助。 （和所有这些堆栈溢出的答案已经帮助我去约会了:)

package test; 

    import java.util.concurrent.DelayQueue; 
    import java.util.concurrent.Delayed; 
    import java.util.concurrent.TimeUnit; 

    public class Test10_DelayQueue { 

     private static final TimeUnit delayUnit = TimeUnit.MILLISECONDS; 
     private static final TimeUnit ripeUnit = TimeUnit.NANOSECONDS; 

     static long startTime; 

     static class Task implements Delayed {  
      public long ripe; 
      public String name;  
      public Task(String name, int delay) { 
      this.name = name; 
      ripe = System.nanoTime() + ripeUnit.convert(delay, delayUnit); 
      } 

     @Override 
     public boolean equals(Object obj) { 
     if (obj instanceof Task) { 
      return compareTo((Task) obj) == 0; 
     } 
     return false; 
     } 

     @Override 
     public int hashCode() { 
     int hash = 7; 
     hash = 67 * hash + (int) (this.ripe^(this.ripe >>> 32)); 
     hash = 67 * hash + (this.name != null ? this.name.hashCode() : 0); 
     return hash; 
     } 

     @Override 
     public int compareTo(Delayed delayed) { 
     if (delayed instanceof Task) { 
      Task that = (Task) delayed; 
      return (int) (this.ripe - that.ripe); 
     } 
     throw new UnsupportedOperationException(); 
     } 

     @Override 
     public long getDelay(TimeUnit unit) { 
     return unit.convert(ripe - System.nanoTime(), ripeUnit); 
     } 

     @Override 
     public String toString() { 
     return "task " + name + " due in " + String.valueOf(getDelay(delayUnit) + "ms"); 
      } 
     } 

     static class TaskAdder implements Runnable { 

     DelayQueue dq; 
     int delay; 

     public TaskAdder(DelayQueue dq, int delay) { 
     this.dq = dq; 
     this.delay = delay; 
     } 

     @Override 
     public void run() { 
     try { 
      Thread.sleep(delay); 

      Task z = new Task("Z", 0); 
      dq.add(z); 

      Long elapsed = System.currentTimeMillis() - startTime; 

      System.out.println("time = " + elapsed + "\tadded " + z); 

     } catch (InterruptedException e) { 
     } 
     } 
    } 

    public static void main(String[] args) { 
     startTime = System.currentTimeMillis(); 
     DelayQueue<Task> taskQ = new DelayQueue<Task>(); 

     Thread thread = new Thread(new TaskAdder(taskQ, 3000)); 
     thread.start(); 

     taskQ.add(new Task("A", 0)); 
     taskQ.add(new Task("B", 10)); 
     taskQ.add(new Task("C", 100)); 
     taskQ.add(new Task("D", 1000)); 
     taskQ.add(new Task("E", 10000)); 
     taskQ.add(new Task("F", 100000)); 

     System.out.println("------initial tasks ---------------"); 
     Task[] tasks = taskQ.toArray(new Task[0]); 
     for (int i = 0; i < tasks.length; i++) { 
     System.out.println(tasks[i]); 
     } 

     System.out.println("------processing--------------------"); 
     try { 
     Long elapsed = System.currentTimeMillis() - startTime; 
     while (elapsed < 15000) { 
      Task task = taskQ.poll(1, TimeUnit.SECONDS); 
      elapsed = System.currentTimeMillis() - startTime; 
      if (task != null) { 
       System.out.println("time = " + elapsed + "\t" + task); 
      } 
     } 

     System.out.println("------remaining after " + elapsed + "ms -----------"); 
     tasks = taskQ.toArray(new Task[0]); 
     for (int i = 0; i < tasks.length; i++) { 
      System.out.println(tasks[i]); 
     } 

     } catch (InterruptedException e) { 
     } 
    } 
    } 

Answer 1

因为你comapareTo方法是漏洞百出。正确的实施如下。一旦你改变如下所有问题将得到解决。总是试图重用compareTo方法，如果还是坚持compareTo合同

return Long.valueOf(this.ripe).compareTo(that.ripe); 

Answer 2

的原因是由于数字溢出。

你compareTo()方法铸造在纳秒long差异int，但超过22秒钟的纳秒不能在int举行，你会得到一个溢出 - 让更多或更少的随机结果，所以如果将来队列中的订单超过2.2秒，则队列中的订单可能会落后于某个队列。

poll()不会超出队列中的下一个项目，当项目放入队列时，其顺序由compareTo方法定义。

此外，equals()应与hashCode()同意，以及compareTo()。有关详细信息，请参阅javadoc for hashCode()。

Answer 3

除非这是实施事件调度程序的练习，否则最好使用ScheduledExecutorService。它会做你想做的所有事情，以及更多。