import java.util.concurrent.*;
 
public class OrderCancelService {
    private static final DelayQueue<OrderTask> delayQueue = new DelayQueue<>();
 
    public static void main(String[] args) throws InterruptedException {
        // 啟動(dòng)消費(fèi)者線(xiàn)程
        new Thread(() -> {
            while (true) {
                try {
                    OrderTask task = delayQueue.take(); // 獲取到期任務(wù)
                    System.out.println("取消訂單：" + task.getOrderId());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }).start();
 
        // 模擬訂單創(chuàng)建
        for (int i = 1; i <= 5; i++) {
            delayQueue.put(new OrderTask(i, System.currentTimeMillis() + 5000)); // 5秒后取消
            System.out.println("訂單" + i + "已創(chuàng)建");
        }
    }
 
    static class OrderTask implements Delayed {
        private final long expireTime;
        private final int orderId;
 
        public OrderTask(int orderId, long expireTime) {
            this.orderId = orderId;
            this.expireTime = expireTime;
        }
 
        public int getOrderId() {
            return orderId;
        }
 
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
 
        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.expireTime, ((OrderTask) o).expireTime);
        }
    }
}

public void cancelExpiredOrders() {
    String sql = "UPDATE orders SET status = 'CANCELLED' WHERE status = 'PENDING' AND create_time < ?";
    try (Connection conn = dataSource.getConnection();
         PreparedStatement ps = conn.prepareStatement(sql)) {
        ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - 30 * 60 * 1000)); // 30分鐘未支付取消
        int affectedRows = ps.executeUpdate();
        System.out.println("取消訂單數(shù)量：" + affectedRows);
    } catch (SQLException e) {
        e.printStackTrace();
    }
}

public void addOrderToQueue(String orderId, long expireTime) {
    jedis.zadd("order_delay_queue", expireTime, orderId);
}
 
public void processExpiredOrders() {
    long now = System.currentTimeMillis();
    Set<String> expiredOrders = jedis.zrangeByScore("order_delay_queue", 0, now);
    for (String orderId : expiredOrders) {
        System.out.println("取消訂單：" + orderId);
        jedis.zrem("order_delay_queue", orderId); // 刪除已處理的訂單
    }
}

public void setOrderWithExpiration(String orderId, long expireSeconds) {
    jedis.setex("order:" + orderId, expireSeconds, "PENDING");
}

public void subscribeToExpirationEvents() {
    Jedis jedis = new Jedis("localhost");
    jedis.psubscribe(new JedisPubSub() {
        @Override
        public void onPMessage(String pattern, String channel, String message) {
            if (channel.equals("__keyevent@0__:expired")) {
                System.out.println("接收到過(guò)期事件，取消訂單：" + message);
                // 執(zhí)行取消訂單的業(yè)務(wù)邏輯
            }
        }
    }, "__keyevent@0__:expired"); // 訂閱過(guò)期事件
}

notify-keyspace-events Ex

public void sendOrderToDelayQueue(String orderId, long delay) {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    ConnectionFactory factory = new ConnectionFactory();
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {
        channel.exchangeDeclare("delayed_exchange", "x-delayed-message", true, false, args);
        channel.queueDeclare("delay_queue", true, false, false, null);
        channel.queueBind("delay_queue", "delayed_exchange", "order.cancel");
 
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(Map.of("x-delay", delay)) // 延遲時(shí)間
                .build();
        channel.basicPublish("delayed_exchange", "order.cancel", props, orderId.getBytes());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Scheduled(cron = "0 */5 * * * ?")
public void scanAndCancelOrders() {
    System.out.println("開(kāi)始掃描并取消過(guò)期訂單");
    // 這里調(diào)用數(shù)據(jù)庫(kù)更新邏輯
}

DataStream<OrderEvent> orderStream = env.fromCollection(orderEvents);
 
orderStream
    .keyBy(OrderEvent::getOrderId)
    .process(new KeyedProcessFunction<String, OrderEvent, Void>() {
        @Override
        public void processElement(OrderEvent event, Context ctx, Collector<Void> out) throws Exception {
            // 注冊(cè)一個(gè)定時(shí)器
            ctx.timerService().registerProcessingTimeTimer(event.getTimestamp() + 30000); // 30秒超時(shí)
        }
 
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Void> out) throws Exception {
            // 定時(shí)器觸發(fā)，執(zhí)行訂單取消邏輯
            System.out.println("訂單超時(shí)取消，訂單ID：" + ctx.getCurrentKey());
        }
    });