Java秒杀系统(十)：RabbitMQ死信队列处理超时未支付的订单

作者： 修罗debug
版权声明：本文为博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。

摘要：本篇博文是“Java秒杀系统实战系列文章”的第十篇，本篇博文我们将采用RabbitMQ的死信队列的方式处理“用户秒杀成功生成订单后，却迟迟没有支付”的情况，一起来见识一下RabbitMQ死信队列在实际业务环境下的强大之处！

内容：对于消息中间件RabbitMQ，Debug其实在前面的篇章中已经简单分享介绍过了，在这里就不再赘述了！在本文我们将采用RabbitMQ的死信队列实现这样的业务需求：“用户在秒杀成功并成功创建一笔订单记录后，理论上应该是执行去支付的操作，但是却存在着一种情况是用户迟迟不肯去支付~至于原因，不得而知！”

对于这种场景，各位小伙伴可以在一些商城平台体验一下，即挑选完商品，加入购物车后，点击去结算，这个时候会有个倒计时，提醒你需要在指定的时间内完成付款，否则订单将失效！

对于这种业务逻辑的处理，传统的做法是采用“定时器的方式”，定时轮询获取已经超过指定时间的订单，然后执行一系列的处理措施（比如再争取给用户发送短信，提醒超过多长时间订单就要失效了等等。。。），在这个秒杀系统中，我们将借助RabbitMQ死信队列这一组件，对该订单执行“失效”的措施！

“死信队列”，顾明思议，是可以延时、延迟一定的时间再处理消息的一种特殊队列，它相对于“普通的队列”而言，可以实现“进入死信队列的消息不立即处理，而是可以等待一定的时间再进行处理”的功能！而普通的队列则不行，即进入队列后的消息会立即被对应的消费者监听消费，如下图所示为普通队列的基本消息模型：

而对于“死信队列”，它的构成以及使用相对而言比较复杂一点，在正常情况，死信队列由三大核心组件组成：死信交换机+死信路由+TTL（消息存活时间~非必需的），而死信队列又可以由“面向生产者的基本交换机+基本路由”绑定而成，故而生产者首先是将消息发送至“基本交换机+基本路由”所绑定而成的消息模型中，即间接性地进入到死信队列中，当过了TTL，消息将“挂掉”，从而进入下一个中转站，即“面下那个消费者的死信交换机+死信路由”所绑定而成的消息模型中。如下图所示：  

下面，我们以实际的代码来构建死信队列的消息模型，并将此消息模型应用到秒杀系统的上述功能模块中。

(1) 首先，需要在RabbitmqConfig配置类创建死信队列的消息模型，其完整的源代码如下所示：

//构建秒杀成功之后-订单超时未支付的死信队列消息模型

@Bean
public Queue successKillDeadQueue(){
    Map<String, Object> argsMap= Maps.newHashMap();
    argsMap.put("x-dead-letter-exchange",env.getProperty("mq.kill.item.success.kill.dead.exchange"));
    argsMap.put("x-dead-letter-routing-key",env.getProperty("mq.kill.item.success.kill.dead.routing.key"));
    return new Queue(env.getProperty("mq.kill.item.success.kill.dead.queue"),true,false,false,argsMap);
}

//基本交换机
@Bean
public TopicExchange successKillDeadProdExchange(){
    return new TopicExchange(env.getProperty("mq.kill.item.success.kill.dead.prod.exchange"),true,false);
}
//创建基本交换机+基本路由 -> 死信队列 的绑定
@Bean
public Binding successKillDeadProdBinding(){
    return BindingBuilder.bind(successKillDeadQueue()).to(successKillDeadProdExchange()).with(env.getProperty("mq.kill.item.success.kill.dead.prod.routing.key"));
}
//真正的队列
@Bean
public Queue successKillRealQueue(){
    return new Queue(env.getProperty("mq.kill.item.success.kill.dead.real.queue"),true);
}
//死信交换机
@Bean
public TopicExchange successKillDeadExchange(){
    return new TopicExchange(env.getProperty("mq.kill.item.success.kill.dead.exchange"),true,false);
}
//死信交换机+死信路由->真正队列 的绑定
@Bean
public Binding successKillDeadBinding(){
    return BindingBuilder.bind(successKillRealQueue()).to(successKillDeadExchange()).with(env.getProperty("mq.kill.item.success.kill.dead.routing.key"));
}

其中，环境变量对象实例env读取的变量是配置在application.properties配置文件中的，取值如下所示：  

#订单超时未支付自动失效-死信队列消息模型
mq.kill.item.success.kill.dead.queue=${mq.env}.kill.item.success.kill.dead.queue
mq.kill.item.success.kill.dead.exchange=${mq.env}.kill.item.success.kill.dead.exchange
mq.kill.item.success.kill.dead.routing.key=${mq.env}.kill.item.success.kill.dead.routing.key

mq.kill.item.success.kill.dead.real.queue=${mq.env}.kill.item.success.kill.dead.real.queue
mq.kill.item.success.kill.dead.prod.exchange=${mq.env}.kill.item.success.kill.dead.prod.exchange
mq.kill.item.success.kill.dead.prod.routing.key=${mq.env}.kill.item.success.kill.dead.prod.routing.key

#单位为ms
mq.kill.item.success.kill.expire=20000

(2) 成功创建了消息模型之后，紧接着，我们需要在通用的RabbitMQ发送消息服务类RabbitSenderService中开发“发送消息入死信队列”的功能，在该功能方法中，我们指定了消息的存活时间TTL，取值为配置的变量：mq.kill.item.success.kill.expire 的值，即20s；其完整的源代码如下所示：  

//秒杀成功后生成抢购订单-发送信息入死信队列，等待着一定时间失效超时未支付的订单
public void sendKillSuccessOrderExpireMsg(final String orderCode){
    try {
        if (StringUtils.isNotBlank(orderCode)){
            KillSuccessUserInfo info=itemKillSuccessMapper.selectByCode(orderCode);
            if (info!=null){
                rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
                rabbitTemplate.setExchange(env.getProperty("mq.kill.item.success.kill.dead.prod.exchange"));
                rabbitTemplate.setRoutingKey(env.getProperty("mq.kill.item.success.kill.dead.prod.routing.key"));
                rabbitTemplate.convertAndSend(info, new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        MessageProperties mp=message.getMessageProperties();
                        mp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        mp.setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME,KillSuccessUserInfo.class);

                        //TODO：动态设置TTL(为了测试方便，暂且设置20s)
                        mp.setExpiration(env.getProperty("mq.kill.item.success.kill.expire"));
                        return message;
                    }
                });
            }
        }
    }catch (Exception e){
        log.error("秒杀成功后生成抢购订单-发送信息入死信队列，等待着一定时间失效超时未支付的订单-发生异常，消息为：{}",orderCode,e.fillInStackTrace());
    }
}

从该“发送消息入死信队列”的代码中，我们可以看到，消息首先是先入到“基本交换机+基本路由”所绑定的死信队列的消息模型中的！当消息到了TTL，自然会从死信队列中出来（即“解脱了”），然后进入下一个中转站，即：“死信交换机+死信路由” 所绑定而成的真正队列的消息模型中，最终真正被消费者监听消费！

此时，可以将整个项目、系统运行在外置的tomcat服务器中，然后打开RabbitMQ后端控制台应用，找到该死信队列，可以看到该死信队列的详细信息，如下图所示：

(3) 最后，是需要在RabbitMQ通用的消息监听服务类RabbitReceiverService 中监听“真正队列”中的消息并进行处理：在这里我们是对该订单进行失效处理（前提是还没付款的情况下！），其完整的源代码如下所示：  

//用户秒杀成功后超时未支付-监听者
@RabbitListener(queues = {"${mq.kill.item.success.kill.dead.real.queue}"},containerFactory = "singleListenerContainer")
public void consumeExpireOrder(KillSuccessUserInfo info){
    try {
        log.info("用户秒杀成功后超时未支付-监听者-接收消息:{}",info);

        if (info!=null){
            ItemKillSuccess entity=itemKillSuccessMapper.selectByPrimaryKey(info.getCode());
            if (entity!=null && entity.getStatus().intValue()==0){
                itemKillSuccessMapper.expireOrder(info.getCode());
            }
        }
    }catch (Exception e){
        log.error("用户秒杀成功后超时未支付-监听者-发生异常：",e.fillInStackTrace());
    }
}

其中，失效更新订单的记录的操作由 itemKillSuccessMapper.expireOrder(info.getCode()); 来实现，其对应的动态Sql的写法如下所示：  

<!--失效更新订单信息-->
<update id="expireOrder">
  UPDATE item_kill_success
  SET status = -1
  WHERE code = #{code} AND status = 0
</update>

(4) 至此，关于RabbitMQ死信队列消息模型的代码实战已经完毕了！最后我只需要在“用户秒杀成功创建订单的那一刻，发送消息入死信队列”的地方调用即可，其调用代码如下所示：  

/**
 * 通用的方法-记录用户秒杀成功后生成的订单-并进行异步邮件消息的通知
 * @param kill
 * @param userId
 * @throws Exception
 */
private void commonRecordKillSuccessInfo(ItemKill kill, Integer userId) throws Exception{
    //TODO:记录抢购成功后生成的秒杀订单记录

    ItemKillSuccess entity=new ItemKillSuccess();
    String orderNo=String.valueOf(snowFlake.nextId());

    //entity.setCode(RandomUtil.generateOrderCode());   //传统时间戳+N位随机数
    entity.setCode(orderNo); //雪花算法
    entity.setItemId(kill.getItemId());
    entity.setKillId(kill.getId());
    entity.setUserId(userId.toString());
    entity.setStatus(SysConstant.OrderStatus.SuccessNotPayed.getCode().byteValue());
    entity.setCreateTime(DateTime.now().toDate());
    //TODO:学以致用，举一反三 -> 仿照单例模式的双重检验锁写法
    if (itemKillSuccessMapper.countByKillUserId(kill.getId(),userId) <= 0){
        int res=itemKillSuccessMapper.insertSelective(entity);

        if (res>0){
            //TODO:进行异步邮件消息的通知=rabbitmq+mail
            rabbitSenderService.sendKillSuccessEmailMsg(orderNo);

            //TODO:入死信队列，用于 “失效” 超过指定的TTL时间时仍然未支付的订单
            rabbitSenderService.sendKillSuccessOrderExpireMsg(orderNo);
        }
    }
}

最后，是进行自测：点击“抢购”按钮，用户秒杀成功后，会发送一条消息入死信队列（这一点可以在RabbitMQ后端控制台中可以看到一条正Ready好的消息），等待20s，即可看到消息转移到真正的队列，并被真正的消费者监听消费，如下所示：  

好了，关于“RabbitMQ死信队列”的介绍以及应用实战本文就暂且介绍到这里了，此种方式可以很灵活对“超时未支付的订单”，进行很好的处理，而且整个过程是“自动、自然”的，而无需人为去手动点击按钮触发了！当然啦，万事万物都并非十全十美的，死信队列也是如此，在一篇文章中我们将介绍此种方式的瑕疵之处，并采用相应的解决方案进行处理！  

补充：

1、目前，这一秒杀系统的整体构建与代码实战已经全部完成了，该秒杀系统对应的视频教程的链接地址为：https://www.fightjava.com/web/index/course/detail/6，可以点击链接进行试看以及学习，实战期间有任何问题都可以留言或者与Debug联系、交流！

2、另外，Debug也开源了该秒杀系统对应的完整的源代码以及数据库，其地址可以来这里下载：https://gitee.com/steadyjack/SpringBoot-SecondKill 记得Fork跟Star啊！！！