springboot+redis实现订单过期(超时取消)功能的方法详解

@configuration
public class rediskeyexpirationconfig {
    
    @bean
    public redismessagelistenercontainer redismessagelistenercontainer(
            redisconnectionfactory connectionfactory) {
        redismessagelistenercontainer container = new redismessagelistenercontainer();
        container.setconnectionfactory(connectionfactory);
        return container;
    }
}

@component
@slf4j
public class orderexpirationlistener {
    
    private static final string order_key_prefix = "order:";
    private static final string order_expire_key_prefix = "order:expire:";
    
    @autowired
    private redistemplate<string, object> redistemplate;
    
    @autowired
    private orderservice orderservice;
    
    /**
     * 监听所有键过期事件
     */
    @eventlistener
    public void handlekeyexpiredevent(keyexpiredevent<string> event) {
        string expiredkey = new string(event.getsource());
        
        if (expiredkey.startswith(order_expire_key_prefix)) {
            string orderid = expiredkey.substring(order_expire_key_prefix.length());
            handleorderexpired(orderid);
        }
    }
    
    private void handleorderexpired(string orderid) {
        log.info("检测到订单过期: {}", orderid);
        
        try {
            // 异步处理，避免阻塞redis监听线程
            completablefuture.runasync(() -> {
                boolean result = orderservice.cancelexpiredorder(orderid);
                if (result) {
                    log.info("订单 {} 已成功取消", orderid);
                } else {
                    log.warn("订单 {} 取消失败或已处理", orderid);
                }
            });
        } catch (exception e) {
            log.error("处理订单过期异常: {}", orderid, e);
        }
    }
}

# 开启键空间通知
notify-keyspace-events ex

# 或者开启所有事件
# notify-keyspace-events ake

spring:
  redis:
    host: localhost
    port: 6379
    database: 0
    # 监听键过期事件
    listen-patterns: "__keyevent@*__:expired"

<dependency>
    <groupid>org.redisson</groupid>
    <artifactid>redisson-spring-boot-starter</artifactid>
    <version>3.23.5</version>
</dependency>

@service
@slf4j
public class orderdelayqueueservice {
    
    @autowired
    private redissonclient redissonclient;
    
    @autowired
    private orderservice orderservice;
    
    private static final string delay_queue_name = "order:delay:queue";
    private static final string processing_set = "order:delay:processing";
    
    /**
     * 添加订单到延时队列
     * @param orderid 订单id
     * @param delaytime 延时时间（分钟）
     */
    public void addtodelayqueue(string orderid, long delaytime) {
        rblockingqueue<string> blockingqueue = redissonclient
            .getblockingqueue(delay_queue_name);
        rdelayedqueue<string> delayedqueue = redissonclient
            .getdelayedqueue(blockingqueue);
        
        // 添加到延时队列
        delayedqueue.offer(orderid, delaytime, timeunit.minutes);
        
        log.info("订单 {} 已添加到延时队列，将在 {} 分钟后过期", orderid, delaytime);
    }
    
    /**
     * 启动延时队列消费者
     */
    @postconstruct
    public void startdelayqueueconsumer() {
        new thread(this::consumedelayqueue, "delay-queue-consumer").start();
    }
    
    private void consumedelayqueue() {
        rblockingqueue<string> blockingqueue = redissonclient
            .getblockingqueue(delay_queue_name);
        
        while (true) {
            try {
                // 从队列中取出过期的订单
                string orderid = blockingqueue.take();
                
                // 处理订单过期
                processexpiredorder(orderid);
                
            } catch (interruptedexception e) {
                thread.currentthread().interrupt();
                log.error("延时队列消费线程被中断", e);
                break;
            } catch (exception e) {
                log.error("处理延时队列消息异常", e);
            }
        }
    }
    
    /**
     * 处理过期订单
     */
    private void processexpiredorder(string orderid) {
        rset<string> processingset = redissonclient.getset(processing_set);
        
        // 使用set防止重复处理
        if (processingset.add(orderid)) {
            try {
                boolean result = orderservice.cancelexpiredorder(orderid);
                if (result) {
                    log.info("延时队列：订单 {} 已过期取消", orderid);
                } else {
                    log.info("延时队列：订单 {} 无需处理", orderid);
                }
            } finally {
                processingset.remove(orderid);
            }
        }
    }
}

@service
@slf4j
public class orderexpirezsetservice {
    
    @autowired
    private stringredistemplate redistemplate;
    
    private static final string order_expire_zset = "order:expire:zset";
    private static final string order_processing_set = "order:processing:set";
    
    /**
     * 添加订单到过期集合
     * @param orderid 订单id
     * @param expiretime 过期时间戳
     */
    public void addordertoexpireset(string orderid, long expiretime) {
        redistemplate.opsforzset().add(order_expire_zset, orderid, expiretime);
        log.info("订单 {} 已添加到过期集合，过期时间: {}", orderid, 
                 new date(expiretime));
    }
    
    /**
     * 批量扫描过期订单
     */
    public void scanexpiredorders() {
        long now = system.currenttimemillis();
        
        // 获取已过期的订单
        set<string> expiredorders = redistemplate.opsforzset()
            .rangebyscore(order_expire_zset, 0, now);
        
        if (expiredorders != null && !expiredorders.isempty()) {
            for (string orderid : expiredorders) {
                processexpiredorder(orderid);
            }
            
            // 移除已处理的订单
            redistemplate.opsforzset().removerangebyscore(
                order_expire_zset, 0, now);
        }
    }
    
    /**
     * 定时扫描任务
     */
    @scheduled(fixeddelay = 30000) // 每30秒执行一次
    public void scheduledscan() {
        log.debug("开始扫描过期订单");
        scanexpiredorders();
    }
    
    private void processexpiredorder(string orderid) {
        // 使用setnx防止重复处理
        boolean success = redistemplate.opsforvalue()
            .setifabsent(order_processing_set + ":" + orderid, "1", 5, timeunit.minutes);
        
        if (boolean.true.equals(success)) {
            try {
                // 处理订单过期逻辑
                handleorderexpired(orderid);
            } finally {
                // 清理处理标记
                redistemplate.delete(order_processing_set + ":" + orderid);
            }
        }
    }
    
    private void handleorderexpired(string orderid) {
        // todo: 实现订单过期处理逻辑
        log.info("处理过期订单: {}", orderid);
    }
}

public enum orderstatus {
    
    pending_payment(0, "待支付"),
    paid(1, "已支付"),
    completed(2, "已完成"),
    cancelled(3, "已取消"),
    expired(4, "已过期");
    
    private final int code;
    private final string description;
    
    orderstatus(int code, string description) {
        this.code = code;
        this.description = description;
    }
    
    // getter方法省略
}

@data
@entity
@table(name = "t_order")
public class order {
    
    @id
    @generatedvalue(strategy = generationtype.identity)
    private long id;
    
    @column(unique = true)
    private string orderno;
    
    private long userid;
    private bigdecimal amount;
    
    @enumerated(enumtype.ordinal)
    private orderstatus status = orderstatus.pending_payment;
    
    @column(name = "expire_time")
    private localdatetime expiretime;
    
    @column(name = "create_time")
    private localdatetime createtime = localdatetime.now();
    
    @column(name = "update_time")
    private localdatetime updatetime = localdatetime.now();
    
    /**
     * 判断订单是否已过期
     */
    public boolean isexpired() {
        return localdatetime.now().isafter(expiretime) 
            && status == orderstatus.pending_payment;
    }
}

@service
@slf4j
@transactional
public class orderservice {
    
    @autowired
    private orderrepository orderrepository;
    
    @autowired
    private stringredistemplate redistemplate;
    
    @autowired
    private orderdelayqueueservice delayqueueservice;
    
    @autowired
    private orderexpirezsetservice zsetservice;
    
    @autowired
    private applicationeventpublisher eventpublisher;
    
    private static final string order_lock_prefix = "order:lock:";
    private static final string order_expire_key_prefix = "order:expire:";
    private static final int order_expire_minutes = 30; // 30分钟未支付过期
    
    /**
     * 创建订单
     */
    public order createorder(long userid, bigdecimal amount) {
        order order = new order();
        order.setorderno(generateorderno());
        order.setuserid(userid);
        order.setamount(amount);
        order.setstatus(orderstatus.pending_payment);
        order.setexpiretime(localdatetime.now().plusminutes(order_expire_minutes));
        
        order = orderrepository.save(order);
        
        // 设置redis过期
        setorderexpire(order.getorderno());
        
        // 添加到延时队列
        delayqueueservice.addtodelayqueue(order.getorderno(), order_expire_minutes);
        
        // 添加到zset
        zsetservice.addordertoexpireset(
            order.getorderno(), 
            order.getexpiretime().atzone(zoneid.systemdefault()).toinstant().toepochmilli()
        );
        
        log.info("创建订单成功: {}, 过期时间: {}", order.getorderno(), order.getexpiretime());
        return order;
    }
    
    /**
     * 设置redis键过期
     */
    private void setorderexpire(string orderno) {
        string key = order_expire_key_prefix + orderno;
        string value = string.valueof(system.currenttimemillis());
        
        // 设置30分钟后过期
        redistemplate.opsforvalue().set(
            key, 
            value, 
            order_expire_minutes, 
            timeunit.minutes
        );
        
        // 同时存储订单信息，用于过期时处理
        map<string, string> orderinfo = new hashmap<>();
        orderinfo.put("orderno", orderno);
        orderinfo.put("userid", "1"); // 实际从订单获取
        orderinfo.put("amount", "100.00");
        
        redistemplate.opsforhash().putall(order_key_prefix + orderno, orderinfo);
    }
    
    /**
     * 支付成功处理
     */
    public boolean processpayment(string orderno) {
        // 获取分布式锁
        string lockkey = order_lock_prefix + orderno;
        boolean locked = redistemplate.opsforvalue()
            .setifabsent(lockkey, "1", 10, timeunit.seconds);
        
        if (boolean.false.equals(locked)) {
            throw new runtimeexception("订单处理中，请稍后");
        }
        
        try {
            order order = orderrepository.findbyorderno(orderno)
                .orelsethrow(() -> new runtimeexception("订单不存在"));
            
            // 检查订单状态
            if (order.getstatus() != orderstatus.pending_payment) {
                throw new runtimeexception("订单状态异常: " + order.getstatus());
            }
            
            // 检查是否过期
            if (order.isexpired()) {
                order.setstatus(orderstatus.expired);
                orderrepository.save(order);
                throw new runtimeexception("订单已过期");
            }
            
            // 更新订单状态
            order.setstatus(orderstatus.paid);
            order.setupdatetime(localdatetime.now());
            orderrepository.save(order);
            
            // 移除过期设置
            removeorderexpire(orderno);
            
            // 发布支付成功事件
            eventpublisher.publishevent(new orderpaidevent(this, order));
            
            log.info("订单支付成功: {}", orderno);
            return true;
            
        } finally {
            // 释放锁
            redistemplate.delete(lockkey);
        }
    }
    
    /**
     * 处理过期订单
     */
    public boolean cancelexpiredorder(string orderno) {
        string lockkey = order_lock_prefix + orderno;
        boolean locked = redistemplate.opsforvalue()
            .setifabsent(lockkey, "1", 10, timeunit.seconds);
        
        if (boolean.false.equals(locked)) {
            return false;
        }
        
        try {
            order order = orderrepository.findbyorderno(orderno)
                .orelsethrow(() -> new runtimeexception("订单不存在"));
            
            // 双重检查：订单是否仍为待支付状态
            if (order.getstatus() != orderstatus.pending_payment) {
                log.info("订单 {} 状态已变更为 {}，跳过取消", orderno, order.getstatus());
                return false;
            }
            
            // 检查是否真的过期
            if (!order.isexpired()) {
                log.info("订单 {} 未过期，跳过取消", orderno);
                return false;
            }
            
            // 更新订单状态
            order.setstatus(orderstatus.expired);
            order.setupdatetime(localdatetime.now());
            orderrepository.save(order);
            
            // 释放库存等业务逻辑
            releasestock(order);
            
            // 发送通知
            sendexpirenotification(order);
            
            log.info("订单 {} 已过期取消", orderno);
            return true;
            
        } finally {
            redistemplate.delete(lockkey);
        }
    }
    
    /**
     * 移除订单过期设置
     */
    private void removeorderexpire(string orderno) {
        // 删除过期key
        redistemplate.delete(order_expire_key_prefix + orderno);
        
        // 从延时队列移除
        // 注意：redisson延时队列不支持直接移除，需要其他方式
        
        // 从zset移除
        redistemplate.opsforzset().remove(order_expire_zset, orderno);
        
        // 删除订单缓存
        redistemplate.delete(order_key_prefix + orderno);
    }
    
    /**
     * 生成订单号
     */
    private string generateorderno() {
        // 时间戳 + 随机数
        return "ord" + 
               system.currenttimemillis() + 
               string.format("%06d", threadlocalrandom.current().nextint(1000000));
    }
    
    private void releasestock(order order) {
        // 释放库存逻辑
        log.info("释放订单 {} 的库存", order.getorderno());
    }
    
    private void sendexpirenotification(order order) {
        // 发送通知逻辑
        log.info("发送订单 {} 过期通知", order.getorderno());
    }
}

public class orderpaidevent extends applicationevent {
    
    private final order order;
    
    public orderpaidevent(object source, order order) {
        super(source);
        this.order = order;
    }
    
    public order getorder() {
        return order;
    }
}

@component
@slf4j
public class orderexpiremanager {
    
    @autowired
    private orderservice orderservice;
    
    @autowired
    private stringredistemplate redistemplate;
    
    private static final string order_expire_zset = "order:expire:zset";
    private static final string order_expire_delay_queue = "order:expire:delay:queue";
    
    /**
     * 三级过期检测策略
     */
    public void startexpiremonitor() {
        // 1. redis键过期事件（实时）
        // 2. 定时任务扫描（兜底）
        // 3. 延时队列（精确控制）
        
        new thread(this::monitordelayqueue, "order-expire-monitor").start();
        new thread(this::scheduledscan, "order-expire-scanner").start();
    }
    
    /**
     * 监控延时队列
     */
    private void monitordelayqueue() {
        while (!thread.currentthread().isinterrupted()) {
            try {
                // 从延时队列获取订单
                string orderid = redistemplate.opsforlist()
                    .rightpop(order_expire_delay_queue, 1, timeunit.seconds);
                
                if (orderid != null) {
                    processexpiredorder(orderid);
                }
            } catch (exception e) {
                log.error("监控延时队列异常", e);
            }
        }
    }
    
    /**
     * 定时扫描
     */
    private void scheduledscan() {
        while (!thread.currentthread().isinterrupted()) {
            try {
                scanexpiredorders();
                thread.sleep(30000); // 30秒扫描一次
            } catch (interruptedexception e) {
                thread.currentthread().interrupt();
                break;
            } catch (exception e) {
                log.error("定时扫描异常", e);
            }
        }
    }
    
    /**
     * 扫描过期订单
     */
    private void scanexpiredorders() {
        long now = system.currenttimemillis();
        set<string> expiredorders = redistemplate.opsforzset()
            .rangebyscore(order_expire_zset, 0, now);
        
        if (expiredorders != null) {
            for (string orderid : expiredorders) {
                processexpiredorder(orderid);
            }
            
            // 移除已处理的订单
            redistemplate.opsforzset().removerangebyscore(
                order_expire_zset, 0, now);
        }
    }
    
    /**
     * 处理过期订单
     */
    private void processexpiredorder(string orderid) {
        // 防重处理
        string lockkey = "order:expire:process:" + orderid;
        boolean locked = redistemplate.opsforvalue()
            .setifabsent(lockkey, "1", 5, timeunit.minutes);
        
        if (boolean.true.equals(locked)) {
            try {
                boolean result = orderservice.cancelexpiredorder(orderid);
                if (result) {
                    log.info("成功处理过期订单: {}", orderid);
                }
            } finally {
                redistemplate.delete(lockkey);
            }
        }
    }
}

@configuration
@enablescheduling
public class orderexpireconfig {
    
    @bean
    public redistemplate<string, object> redistemplate(
            redisconnectionfactory connectionfactory) {
        redistemplate<string, object> template = new redistemplate<>();
        template.setconnectionfactory(connectionfactory);
        
        // 设置key和value的序列化方式
        template.setkeyserializer(new stringredisserializer());
        template.setvalueserializer(new genericjackson2jsonredisserializer());
        
        template.sethashkeyserializer(new stringredisserializer());
        template.sethashvalueserializer(new genericjackson2jsonredisserializer());
        
        template.afterpropertiesset();
        return template;
    }
    
    @bean
    public orderexpiremanager orderexpiremanager() {
        return new orderexpiremanager();
    }
    
    @postconstruct
    public void init() {
        // 启动过期监控
        orderexpiremanager().startexpiremonitor();
    }
}

@restcontroller
@requestmapping("/api/orders")
@slf4j
public class ordercontroller {
    
    @autowired
    private orderservice orderservice;
    
    @postmapping("/create")
    public apiresponse<order> createorder(@requestbody createorderrequest request) {
        order order = orderservice.createorder(
            request.getuserid(), 
            request.getamount()
        );
        return apiresponse.success(order);
    }
    
    @postmapping("/{orderno}/pay")
    public apiresponse<void> payorder(@pathvariable string orderno) {
        boolean success = orderservice.processpayment(orderno);
        if (success) {
            return apiresponse.success("支付成功");
        } else {
            return apiresponse.error("支付失败");
        }
    }
    
    @getmapping("/{orderno}/status")
    public apiresponse<orderstatus> getorderstatus(@pathvariable string orderno) {
        // 从redis或数据库获取订单状态
        return apiresponse.success(orderstatus.pending_payment);
    }
}

@springboottest
@slf4j
class orderexpiretest {
    
    @autowired
    private orderservice orderservice;
    
    @autowired
    private stringredistemplate redistemplate;
    
    @test
    void testorderexpire() throws interruptedexception {
        // 创建订单
        order order = orderservice.createorder(1l, new bigdecimal("100.00"));
        
        // 验证redis中已设置过期
        string expirekey = "order:expire:" + order.getorderno();
        string value = redistemplate.opsforvalue().get(expirekey);
        assertnotnull(value);
        
        // 验证订单状态
        assertequals(orderstatus.pending_payment, order.getstatus());
        
        // 等待订单过期
        thread.sleep(2000); // 实际应该等30分钟
        
        // 测试支付
        boolean paid = orderservice.processpayment(order.getorderno());
        asserttrue(paid);
    }
    
    @test
    void testconcurrentpay() throws interruptedexception {
        order order = orderservice.createorder(2l, new bigdecimal("200.00"));
        
        executorservice executor = executors.newfixedthreadpool(5);
        countdownlatch latch = new countdownlatch(5);
        
        atomicinteger successcount = new atomicinteger(0);
        
        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                try {
                    boolean result = orderservice.processpayment(order.getorderno());
                    if (result) {
                        successcount.incrementandget();
                    }
                } catch (exception e) {
                    log.error("支付异常", e);
                } finally {
                    latch.countdown();
                }
            });
        }
        
        latch.await();
        executor.shutdown();
        
        // 只有一个支付成功
        assertequals(1, successcount.get());
    }
}

@component
@slf4j
public class orderexpiremonitor {
    
    @autowired
    private stringredistemplate redistemplate;
    
    @scheduled(fixedrate = 60000) // 每分钟执行一次
    public void monitorexpireorders() {
        string expirekeypattern = "order:expire:*";
        
        // 统计待过期订单数量
        set<string> keys = redistemplate.keys(expirekeypattern);
        long expirecount = keys != null ? keys.size() : 0;
        
        // 统计即将在5分钟内过期的订单
        long soonexpirecount = keys.stream()
            .map(key -> redistemplate.getexpire(key, timeunit.seconds))
            .filter(ttl -> ttl != null && ttl > 0 && ttl <= 300)
            .count();
        
        // 记录监控日志
        log.info("订单过期监控 - 待过期订单: {}, 即将过期订单: {}", 
                 expirecount, soonexpirecount);
        
        // 发送告警
        if (soonexpirecount > 100) {
            sendalert("大量订单即将过期: " + soonexpirecount + " 个");
        }
    }
    
    private void sendalert(string message) {
        // 发送告警到监控系统
        log.warn("订单过期告警: {}", message);
    }
}