// 1. 定義延遲任務(wù)
public class OrderCancelTask implements Delayed {
    private final String orderId;
    private final long executeTime; // 絕對時間戳（毫秒）

    public OrderCancelTask(String orderId, long delayMs) {
        this.orderId = orderId;
        this.executeTime = System.currentTimeMillis() + delayMs;
    }

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

    @Override
    public int compareTo(Delayed o) {
        return Long.compare(this.executeTime, ((OrderCancelTask) o).executeTime);
    }

    public void execute() {
        System.out.println("取消訂單: " + orderId + " at " + LocalDateTime.now());
        // 實際業(yè)務(wù)邏輯：orderService.cancelOrder(orderId);
    }
}

// 2. 啟動消費者線程
@Component
public class DelayQueueConsumer {
    private final DelayQueue<OrderCancelTask> queue = new DelayQueue<>();

    @PostConstruct
    public void startConsumer() {
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    OrderCancelTask task = queue.take(); // 阻塞直到任務(wù)到期
                    task.execute();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "DelayQueue-Consumer").start();
    }

    public void addTask(String orderId, long delayMs) {
        queue.offer(new OrderCancelTask(orderId, delayMs));
    }
}

// 后臺線程：定期從 Redis 加載即將到期的長任務(wù)
@Scheduled(fixedDelay = 30_000)
public void loadLongDelayTasks() {
    long now = System.currentTimeMillis();
    Set<String> tasks = redis.zRangeByScore("delay_tasks", now, now + 60_000);
    tasks.forEach(task -> {
        // 將任務(wù)加入本地時間輪，延遲 = (executeTime - now)
        hashedWheelTimer.newTimeout(timeout -> process(task), delay, TimeUnit.MILLISECONDS);
        redis.zRem("delay_tasks", task);
    });
}

# application.yml
spring:
  redis:
    cluster:
      nodes: 
        - 192.168.1.10:7000
        - 192.168.1.11:7001
        - 192.168.1.12:7002
    timeout: 2000ms
    lettuce:
      pool:
        max-active: 20

// 消費前檢查是否已處理
public void processOrderCancel(String orderId) {
    String key = "delay_task_processed:" + orderId;
    Boolean isProcessed = redis.set(key, "1", SetArgs.Builder.nx().ex(86400));
    
    if (Boolean.TRUE.equals(isProcessed)) {
        // 執(zhí)行業(yè)務(wù)邏輯（取消訂單）
        orderService.cancelOrder(orderId);
    } else {
        log.warn("任務(wù)已處理，跳過重復消費: {}", orderId);
    }
}

// 消費者邏輯（帶重試）
public void consumeWithRetry() throws InterruptedException {
    while (true) {
        OrderTask task = blockingQueue.take();
        try {
            processOrderCancel(task.getOrderId());
        } catch (Exception e) {
            // 重試次數(shù)+1
            int retryCount = task.getRetryCount() + 1;
            if (retryCount <= MAX_RETRY) {
                // 指數(shù)退避：2^retry * 30秒
                long delay = (long) Math.pow(2, retryCount) * 30;
                task.setRetryCount(retryCount);
                redissonUtil.addDelayQueue(task, delay, TimeUnit.SECONDS, QUEUE_NAME);
            } else {
                // 進入死信隊列（人工處理）
                deadLetterQueue.offer(task);
                alertService.sendAlert("延遲任務(wù)失敗: " + task.getOrderId());
            }
        }
    }
}

// 定時監(jiān)控任務(wù)積壓
@Scheduled(fixedRate = 30_000)
public void monitorDelayQueue() {
    Long size = redis.zCard("redisson_delay_queue_timeout:{order_cancel}");
    if (size > 1000) {
        alertService.sendAlert("Redisson 延遲隊列積壓嚴重: " + size);
    }
}

# application.yml
spring:
  rabbitmq:
    addresses: node1:5672,node2:5672,node3:5672
    virtual-host: /
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
    listener:
      simple:
        acknowledge-mode: manual
        concurrency: 5
        max-concurrency: 20

// 發(fā)送延遲消息（30分鐘）
public void sendDelayMessage(String orderId) {
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    
    // 消息確認回調(diào)
    correlationData.getFuture().addCallback(result -> {
        if (result.isAck()) {
            log.info("消息投遞成功: {}", orderId);
        } else {
            // 投遞失敗 → 落庫補償
            delayTaskCompensationService.saveToDb(orderId, 1800);
        }
    }, throwable -> {
        // 異常 → 落庫補償
        delayTaskCompensationService.saveToDb(orderId, 1800);
    });
    
    Message message = MessageBuilder
        .withBody(orderId.getBytes())
        .setExpiration("1800000") // 30分鐘
        .build();
    
    rabbitTemplate.send("delay.exchange", "delay.key", message, correlationData);
}

@RabbitListener(queues = "order.cancel.dlq")
public void handleDelayMessage(Message message, Channel channel) throws IOException {
    String orderId = new String(message.getBody());
    
    try {
        // 冪等檢查
        if (isOrderProcessed(orderId)) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        
        // 執(zhí)行業(yè)務(wù)
        orderService.cancelOrder(orderId);
        
        // 手動 ACK
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        
    } catch (Exception e) {
        // NACK 并重新入隊（最多3次）
        if (message.getMessageProperties().getRedelivered()) {
            // 進入死信隊列（人工處理）
            deadLetterService.save(message);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

// 定時補償任務(wù)（每5分鐘掃描）
@Scheduled(fixedDelay = 300_000)
public void compensateMissedTasks() {
    List<DelayTask> tasks = delayTaskMapper.selectUnprocessedBefore(
        Instant.now().minusSeconds(60) // 查找1分鐘前應執(zhí)行但未處理的任務(wù)
    );
    
    for (DelayTask task : tasks) {
        // 重新發(fā)送到 RabbitMQ
        rabbitMQService.sendDelayMessage(task.getOrderId(), 0); // 立即執(zhí)行
        // 標記為已補償
        delayTaskMapper.markAsCompensated(task.getId());
    }
}