Java DelayQueue延迟队列的原理与应用场景详解

public class DelayedTask implements Delayed {
    private final String taskName;
    private final long executeTime; // 执行时间戳（纳秒）
    private final long delay; // 延迟时间（毫秒）

    public DelayedTask(String taskName, long delayInMillis) {
        this.taskName = taskName;
        this.delay = delayInMillis;
        this.executeTime = System.nanoTime() + 
                          TimeUnit.NANOSECONDS.convert(delayInMillis, TimeUnit.MILLISECONDS);
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long remaining = executeTime - System.nanoTime();
        return unit.convert(remaining, TimeUnit.NANOSECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        if (other == this) return 0;
        long diff = this.getDelay(TimeUnit.NANOSECONDS) - 
                   other.getDelay(TimeUnit.NANOSECONDS);
        return Long.compare(diff, 0);
    }

    @Override
    public String toString() {
        return "Task[" + taskName + "]@" + 
               Instant.ofEpochMilli(TimeUnit.MILLISECONDS.convert(executeTime, TimeUnit.NANOSECONDS));
    }
}

public class DelayQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedTask> queue = new DelayQueue<>();
        
        // 添加延迟任务
        queue.put(new DelayedTask("Task-1", 3000)); // 3秒后执行
        queue.put(new DelayedTask("Task-2", 1000)); // 1秒后执行
        queue.put(new DelayedTask("Task-3", 5000)); // 5秒后执行

        System.out.println("⌛ 开始等待延迟任务...");
        
        // 循环取出到期任务
        while (!queue.isEmpty()) {
            DelayedTask task = queue.take(); // 阻塞直到有任务到期
            System.out.printf("[%s] 执行任务: %s%n", 
                LocalTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME), 
                task);
        }
    }
}

public class OrderCancelSystem {
    private static final DelayQueue<DelayedOrder> cancelQueue = new DelayQueue<>();
    
    // 订单延迟项
    static class DelayedOrder implements Delayed {
        private final String orderId;
        private final long expireTime;
        
        public DelayedOrder(String orderId, long delay, TimeUnit unit) {
            this.orderId = orderId;
            this.expireTime = System.nanoTime() + unit.toNanos(delay);
        }
        
        // 实现Delayed接口方法...
        
        void cancelOrder() {
            System.out.printf("[%s] 订单超时取消: %s%n", 
                LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME), 
                orderId);
            // 实际业务中调用订单取消服务
        }
    }

    // 订单处理器
    static class OrderProcessor extends Thread {
        @Override
        public void run() {
            while (true) {
                try {
                    DelayedOrder order = cancelQueue.take();
                    order.cancelOrder();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    public static void main(String[] args) {
        // 启动订单处理线程
        new OrderProcessor().start();
        
        // 模拟订单创建
        String[] orders = {"ORD-1001", "ORD-1002", "ORD-1003"};
        for (String orderId : orders) {
            cancelQueue.put(new DelayedOrder(orderId, 30, TimeUnit.MINUTES));
            System.out.printf("创建订单: %s @ %s%n", orderId, LocalTime.now());
        }
    }
}

private final PriorityQueue<E> q = new PriorityQueue<>();

private final transient ReentrantLock lock = new ReentrantLock();

private final Condition available = lock.newCondition();

// 错误示范：混合使用单位和时间戳
long delay = 1000; // 这是毫秒还是秒？

// 正确姿势：统一使用TimeUnit
long nanos = TimeUnit.SECONDS.toNanos(5);

public long getDelay(TimeUnit unit) {
    long remaining = executeTime - System.nanoTime();
    // 必须处理负值情况！
    return unit.convert(Math.max(remaining, 0), TimeUnit.NANOSECONDS);
}

// 使用容量限制（Java 7+）
new DelayQueue<>().remainingCapacity(); // 始终返回Integer.MAX_VALUE
// 实际方案：用Semaphore做流量控制

// 使用时间差而非绝对时间
long start = System.nanoTime();
long elapsed = System.nanoTime() - start;

// 使用单调时钟（避免系统时间调整影响）
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);

public void shutdown() {
  Thread.currentThread().interrupt();
  // 清空队列中的待处理任务
  queue.clear(); 
}

// 通过JMX暴露队列大小
@ManagedAttribute
public int getQueueSize() {
  return delayQueue.size();
}

public class TaskScheduler {
  private final DelayQueue<DelayedTask> queue = new DelayQueue<>();
  
  public void schedule(Runnable task, long delay, TimeUnit unit) {
      queue.put(new DelayedTask(task, delay, unit));
  }
}

long preciseDelay = TimeUnit.SECONDS.toNanos(1);
// 在getDelay()中使用：
return unit.convert(nanosRemaining, TimeUnit.NANOSECONDS);