DelayQueue意外行为。DrainTo仅从队列中删除1个过期项。

Question

我希望能够遍历DelayQueue中未过期的元素。 Transaction类实现了Delayed接口，并具有时间戳字段，该字段表示交易在协调世界时（UTC）发起时的时间戳（而非当前时间戳）。

public class Transaction implements Delayed {
    private final Double amount;
    private final Long timestamp;   //timestamp of a time when the item was created and send here


    public Transaction(double amount, long timestamp) {
        this.amount = amount;
        this.timestamp = timestamp;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long delay = unit.convert(ONEMINUTE - (System.currentTimeMillis() - timestamp), TimeUnit.MILLISECONDS);
    return delay;
}

    @Override
    public int compareTo(Delayed delayed) {
        if (delayed == this) {
            return 0;
        }

        if (delayed instanceof Transaction) {
            return 0;
        }

        long diff = (getDelay(TimeUnit.MILLISECONDS) - delayed.getDelay(TimeUnit.MILLISECONDS));
        return ((diff == 0) ? 0 : ((diff < 0) ? -1 : 1));
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + timestamp.hashCode();
        return result;
    }

    @Override
    public boolean equals( Object obj ) {
        if( this == obj ) {
            return true;
        }

        if( obj == null ) {
            return false;
        }

        if( !( obj instanceof Transaction ) ) {
            return false;
        }

        final Transaction other = ( Transaction )obj;
        return timestamp.equals(other.timestamp);
    }
}

下面的TransactionManager类会将传入的事务添加到队列中，如果新来的事务比1分钟还年轻。在getStatistics函数中，旧的事务应该从队列中移除，并且队列应该只包含年轻于1分钟的事务。

public class TransactionManager {

    private DelayQueue<Transaction> transactions;


    public TransactionManager() {
        transactions = new DelayQueue<>();
        System.setProperty("user.timezone", "UTC");
    }

    public Object createTransaction(String json) {
        JSONObject jsonObject = null;
        try {
            jsonObject = JsonValidator.validateTransactionJson(json);
        } catch (Exception ex) {
            return new ResponseEntity(HttpStatus.UNPROCESSABLE_ENTITY);
        }
        long delay = System.currentTimeMillis() - ((Long) jsonObject.get(TIMESTAMP));
        if (delay > ONEMINUTE) {
            return new ResponseEntity(HttpStatus.NO_CONTENT);
        }
        transactions.add(new Transaction((Double) jsonObject.get(AMOUNT), (Long) jsonObject.get(TIMESTAMP)));
        return new ResponseEntity(HttpStatus.OK);
    }

    public long getStatistics() {
        List<Transaction> tempForCleaning = new ArrayList<>();
        transactions.drainTo(tempForCleaning);
        tempForCleaning.clear();
        StatisticJSON statistics = new StatisticJSON();
        transactions.stream().forEach(transaction -> {
            statistics.setCount(statistics.getCount() + 1);
        });
        return statistics.getCount();
    }
}

在这个测试中，我创建了5个距离现在40秒的交易和3个距离现在10秒的交易。因此，在等待45秒后，前5个交易应该被清除，队列应该只包含3个交易。然而，方法drainTo只删除了一个旧的交易。

@Test
public void test() {
    DateTime dateTime = new DateTime(DateTimeZone.UTC);
    long fortyMilliSecondsAgo = dateTime.minusSeconds(40).getMillis();
    long twentyMilliSecondsAgo = dateTime.minusSeconds(10).getMillis();
    for (int i = 0; i < 5; i++) {
        createTransaction(fortyMilliSecondsAgo);
    }
    for (int i = 0; i < 3; i++) {
        createTransaction(twentyMilliSecondsAgo);
    }
    Assert.assertTrue(transactionManager.getStatistics() == 8);
    try {
        TimeUnit.SECONDS.sleep(45);
        System.out.println("\n\n\n");
        Assert.assertTrue(transactionManager.getStatistics() == 3);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

private void createTransaction(long timestamp) {
    transactionManager.createTransaction("{\"amount\":100.0,\"timestamp\":" + timestamp + "}");
}

我觉得有点不对劲，为什么 drainTo 只删除了一个过期的项目，即使还剩下 4 个，但我无法找到原因...

- Marat

Answer 1

if (delayed instanceof Transaction) { return 0; }

看起来不太对 - 如果你想让compareTo与getDelay()保持一致，你应该删除那一部分。因此，你的方法可能应该像这样（使用静态导入）：

public int compareTo(Delayed delayed) {
  return delayed == this
    ? 0
    : Long.compare(getDelay(MILLISECONDS), delayed.getDelay(MILLISECONDS));
}

- assylias

---

谢谢，你说得很对，那部分我没有看到 :) - Marat