延迟队列与优先级消费实现

Kafka 的消费模型：
  - 消息按 offset 严格顺序
  - 消费者只能从 offset 顺序推进
  - 无法「跳到」某条特定消息

// 消费者 poll 到消息后，判断是否到期
for (ConsumerRecord<String, Order> r : records) {
    Order o = r.value();
    if (o.deliverAt <= System.currentTimeMillis()) {
        process(o);            // 到期，处理
    } else {
        // 未到期：不提交 offset，等待下次 poll 重读
        break;                 // 注意：会阻塞后续消息
    }
}

orders.delay（延迟消息） → 消费者读 → 未到期 → consumer.pause(partitions)
                                     → 到期   → consumer.resume(partitions)

// 用 pause/resume 控制可见性
consumer.pause(consumer.assignment());
while (running) {
    ConsumerRecords<String,Order> rs = consumer.poll(Duration.ofMillis(100));
    long now = System.currentTimeMillis();
    for (ConsumerRecord<String,Order> r : rs) {
        if (r.value().deliverAt > now) {
            long waitMs = r.value().deliverAt - now;
            consumer.seek(r.topicPartition(), r.offset());   // 回退
            consumer.pause(List.of(r.topicPartition()));
            scheduler.schedule(() -> consumer.resume(List.of(r.topicPartition())),
                               waitMs, TimeUnit.MILLISECONDS);
            break;
        }
        process(r.value());
    }
    consumer.commitSync();
}

orders.delay.1m   延迟 1 分钟级
orders.delay.5m   延迟 5 分钟级
orders.delay.30m  延迟 30 分钟级
orders.delay.1h   延迟 1 小时级

消息进入最近的、不大于目标延迟的层级
  - 目标延迟 32 分钟 → 进 orders.delay.30m
消费者读到后：
  - 剩余延迟 > 0 → 投递到「更小一级」的 topic
  - 剩余延迟 <= 0 → 投递回主 topic，正式处理

// 30m 层消费者
long remaining = r.value().deliverAt - System.currentTimeMillis();
if (remaining <= 0) {
    producer.send(new ProducerRecord<>("orders", r.key(), r.value()));  // 到期，回主 topic
} else if (remaining <= Duration.ofMinutes(5).toMillis()) {
    producer.send(new ProducerRecord<>("orders.delay.1m", r.key(), r.value()));
} else {
    producer.send(new ProducerRecord<>("orders.delay.5m", r.key(), r.value()));
}

① 生产：延迟消息写入 Redis ZSet，score = deliver_at
② 调度：定时任务扫描 ZSet 中 score <= now 的成员
③ 投递：到期成员投递到 Kafka 主 topic
④ 处理：普通消费者顺序处理

// Redis ZSet 调度
String key = "delay:orders";
// 写入：score = 到期时间戳
jedis.zadd(key, deliverAt, payloadJson);

// 调度线程：拉取到期的
Set<String> due = jedis.zrangeByScore(key, 0, System.currentTimeMillis(), 0, 100);
for (String payload : due) {
    if (jedis.zrem(key, payload) == 1) {     // 原子摘除，防重复
        producer.send(new ProducerRecord<>("orders", payload));
    }
}

orders.p0   高优先级（VIP、告警）
orders.p1   中优先级（普通业务）
orders.p2   低优先级（日志、统计）

// 权重：p0 : p1 : p2 = 5 : 3 : 1
int[] weights = {5, 3, 1};
String[] topics = {"orders.p0", "orders.p1", "orders.p2"};
int[] counters = {0, 0, 0};

while (running) {
    for (int i = 0; i < topics.length; i++) {
        if (counters[i] < weights[i]) {
            ConsumerRecords<String,String> rs = consumer.poll(topic(topicIndex(topics[i])));
            for (var r : rs) process(r);
            counters[i]++;
        }
    }
    if (allSaturated()) Arrays.fill(counters, 0);   // 重置配额
}

// 消费者 poll → 入优先队列 → 按优先级出队处理
PriorityQueue<ConsumerRecord<String,String>> pq =
    new PriorityQueue<>(Comparator.comparingInt(r -> priority(r)));

// poll 线程：入队
for (var r : consumer.poll(Duration.ofMillis(100))) pq.offer(r);
// 处理线程：出队
while (running) {
    ConsumerRecord<String,String> r = pq.poll();
    if (r != null) process(r);
}

orders.p0 → group-vip（独享线程池 + 独享消费者）
orders.p1 → group-normal
orders.p2 → group-batch

维度组合：
  延迟档位（30m / 5m / 1m）× 优先级（p0 / p1 / p2）
  → 9 个 topic？成本爆炸

orders.delay.30m / 5m / 1m    ← 承载延迟（3 个 topic）
每个 topic 内消息带 priority header  ← 承载优先级
消费者：读到后按 priority 路由到不同处理队列

延迟队列：各档位 topic 的 Lag、到期消息处理延迟分布
优先级：各优先级 topic 的消费速率、低优先级饿死时长
组合：端到端延迟（生产 → 到期 → 处理完成）