《Java 并发编程·线程与内存模型》11.2 延迟队列与时间轮

private final transient java.util.concurrent.locks.ReentrantLock lock;
private final java.util.PriorityQueue<E> q;
private java.lang.Thread leader;
private final java.util.concurrent.locks.Condition available;

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            E first = q.peek();
            if (first == null)
                available.await();                 // 队列空，无限等
            else {
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0L)
                    return q.poll();               // 已到期，出队
                first = null;                      // 等的时候不持有元素引用
                if (leader != null)
                    available.await();             // 已有人定时等，我无限等
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);   // 只有 leader 定时等
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

static final class RetryJob implements Delayed {
    final String id;
    final long readyAtNanos;
    final int attempt;

    RetryJob(String id, long delayMs, int attempt) {
        this.id = id;
        this.readyAtNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delayMs);
        this.attempt = attempt;
    }

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

    @Override
    public int compareTo(Delayed other) {
        return Long.compare(this.readyAtNanos, ((RetryJob) other).readyAtNanos);
    }
}

while (true) {
    RetryJob job = q.take();
    boolean ok = job.attempt == maxAttempt - 1;
    if (ok) break;
    long backoff = 50L << job.attempt;        // 50, 100, 200...
    q.put(new RetryJob(job.id, backoff, job.attempt + 1));
}

== 实验二：失败重试 + 指数退避 ==
  +   0ms  尝试 loan-9001#0 -> 失败
         退避 50ms 后重试
  +  60ms  尝试 loan-9001#1 -> 失败
         退避 100ms 后重试
  + 173ms  尝试 loan-9001#2 -> 成功
  重试期间队列长度最终 = 0

ticks   = ceil(d / tickDuration)
dueTick = currentTick + ticks
slot    = dueTick % wheelSize
rounds  = dueTick / wheelSize - currentTick / wheelSize

void add(String name, long delayMs, long t0) {
    long ticks = (delayMs + TICK_MS - 1) / TICK_MS;   // 向上取整到 tick
    long dueTick = currentTick + ticks;
    int slot = (int) (dueTick % SLOTS);
    wheel.get(slot).add(new Job(name, dueTick, t0 + TimeUnit.MILLISECONDS.toNanos(delayMs)));
}

void awaitNextTick(long t0) {
    long nextAt = t0 + TimeUnit.MILLISECONDS.toNanos(TICK_MS * (currentTick + 1));
    long remain;
    while ((remain = nextAt - System.nanoTime()) > 0) {
        LockSupport.parkNanos(remain);   // 睡到「下一个绝对刻度」，不是固定睡 50ms
    }
    currentTick++;
}

== 第二版：按绝对刻度推进（两次运行） ==
  w0   槽 1 约定 + 30ms，实际 + 62ms / +54ms，误差 32ms / 24ms
  w1   槽 2 约定 + 70ms，实际 +100ms / +100ms，误差 30ms / 30ms
  w2   槽 3 约定 +120ms，实际 +153ms / +151ms，误差 33ms / 31ms
  w3   槽 6 约定 +260ms，实际 +310ms / +305ms，误差 50ms / 45ms
  w4   槽 0 约定 +380ms，实际 +408ms / +402ms，误差 28ms / 22ms
  w5   槽 1 约定 +430ms，实际 +457ms / +456ms，误差 27ms / 26ms
  平均量化误差 = 33.3ms / 29.7ms

== 同一组延迟交给 DelayQueue ==
  d0   约定  30ms，实际 + 30ms，误差 0ms
  d1   约定  70ms，实际 + 74ms，误差 4ms
  d2   约定 120ms，实际 +122ms，误差 2ms
  d3   约定 260ms，实际 +261ms，误差 1ms
  d4   约定 380ms，实际 +382ms，误差 2ms
  d5   约定 430ms，实际 +434ms，误差 4ms
  平均误差 = 2.2ms

public final class NoticeRetryQueue implements AutoCloseable {
    private static final int MAX_ATTEMPT = 3;

    private final DelayQueue<Retry> queue = new DelayQueue<>();
    private final Consumer<Loan> deliver;
    private final Thread worker;

    public NoticeRetryQueue(Consumer<Loan> deliver) {
        this.deliver = deliver;
        this.worker = Thread.ofPlatform().name("notice-retry").daemon(true).start(this::loop);
    }

    public void retryLater(Loan loan, int attempt) {
        queue.put(Retry.after(loan, attempt, 200L << attempt));   // 200, 400, 800ms
    }

    private void loop() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                Retry r = queue.poll(500, TimeUnit.MILLISECONDS);
                if (r == null) continue;                 // 空转时定期醒来，便于关闭
                try {
                    deliver.accept(r.loan());
                } catch (RuntimeException e) {
                    if (r.attempt() + 1 < MAX_ATTEMPT) {
                        retryLater(r.loan(), r.attempt() + 1);
                    } else {
                        System.err.println("notice dropped: " + r.loan().loanId());
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void close() {
        worker.interrupt();
    }

    private record Retry(Loan loan, int attempt, long readyAtNanos) implements Delayed {
        static Retry after(Loan loan, int attempt, long backoffMs) {
            return new Retry(loan, attempt, System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(backoffMs));
        }

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

        @Override public int compareTo(Delayed o) {
            return Long.compare(readyAtNanos, ((Retry) o).readyAtNanos);
        }
    }
}