Java秒杀系统实战系列~RabbitMQ死信队列处理超时未支付的订单

简介: 摘要:本篇博文是“Java秒杀系统实战系列文章”的第十篇,本篇博文我们将采用RabbitMQ的死信队列的方式处理“用户秒杀成功生成订单后,却迟迟没有支付”的情况,一起来见识一下RabbitMQ死信队列在实际业务环境下的强大之处!内容:对于消息中间件RabbitMQ,Debug其实在前面的篇章中已...

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

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

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

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

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

image

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

image

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

(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后端控制台应用,找到该死信队列,可以看到该死信队列的详细信息,如下图所示:

image

(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,即可看到消息转移到真正的队列,并被真正的消费者监听消费,如下所示:

image

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

补充:

1、由于相应的博客的更新可能并不会很快,故而如果有想要快速入门以及实战整套系统的,可以参考阅读: Java商城秒杀系统的设计与实战视频教程(SpringBoot版)

2、目前,这一秒杀系统的整体构建与代码实战已经全部完成了,完整的源代码数据库地址可以来这里下载:https://gitee.com/steadyjack/SpringBoot-SecondKill

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
1月前
|
监控 Java API
如何使用Java语言快速开发一套智慧工地系统
使用Java开发智慧工地系统,采用Spring Cloud微服务架构和前后端分离设计,结合MySQL、MongoDB数据库及RESTful API,集成人脸识别、视频监控、设备与环境监测等功能模块,运用Spark/Flink处理大数据,ECharts/AntV G2实现数据可视化,确保系统安全与性能,采用敏捷开发模式,提供详尽文档与用户培训,支持云部署与容器化管理,快速构建高效、灵活的智慧工地解决方案。
|
21天前
|
设计模式 消息中间件 搜索推荐
Java 设计模式——观察者模式:从优衣库不使用新疆棉事件看系统的动态响应
【11月更文挑战第17天】观察者模式是一种行为设计模式,定义了一对多的依赖关系,使多个观察者对象能直接监听并响应某一主题对象的状态变化。本文介绍了观察者模式的基本概念、商业系统中的应用实例,如优衣库事件中各相关方的动态响应,以及模式的优势和实际系统设计中的应用建议,包括事件驱动架构和消息队列的使用。
|
29天前
|
消息中间件 存储 Java
RocketMQ文件刷盘机制深度解析与Java模拟实现
【11月更文挑战第22天】在现代分布式系统中,消息队列(Message Queue, MQ)作为一种重要的中间件,扮演着连接不同服务、实现异步通信和消息解耦的关键角色。Apache RocketMQ作为一款高性能的分布式消息中间件,广泛应用于实时数据流处理、日志流处理等场景。为了保证消息的可靠性,RocketMQ引入了一种称为“刷盘”的机制,将消息从内存写入到磁盘中,确保消息持久化。本文将从底层原理、业务场景、概念、功能点等方面深入解析RocketMQ的文件刷盘机制,并使用Java模拟实现类似的功能。
40 3
|
2月前
|
存储 消息中间件 安全
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
【10月更文挑战第9天】本文介绍了如何利用JUC组件实现Java服务与硬件通过MQTT的同步通信(RRPC)。通过模拟MQTT通信流程,使用`LinkedBlockingQueue`作为消息队列,详细讲解了消息发送、接收及响应的同步处理机制,包括任务超时处理和内存泄漏的预防措施。文中还提供了具体的类设计和方法实现,帮助理解同步通信的内部工作原理。
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
|
1月前
|
运维 自然语言处理 供应链
Java云HIS医院管理系统源码 病案管理、医保业务、门诊、住院、电子病历编辑器
通过门诊的申请,或者直接住院登记,通过”护士工作站“分配患者,完成后,进入医生患者列表,医生对应开具”长期医嘱“和”临时医嘱“,并在电子病历中,记录病情。病人出院时,停止长期医嘱,开具出院医嘱。进入出院审核,审核医嘱与住院通过后,病人结清缴费,完成出院。
90 3
|
1月前
|
Java 数据库连接 数据库
深入探讨Java连接池技术如何通过复用数据库连接、减少连接建立和断开的开销,从而显著提升系统性能
在Java应用开发中,数据库操作常成为性能瓶颈。本文通过问题解答形式,深入探讨Java连接池技术如何通过复用数据库连接、减少连接建立和断开的开销,从而显著提升系统性能。文章介绍了连接池的优势、选择和使用方法,以及优化配置的技巧。
39 1
|
1月前
|
JavaScript Java 项目管理
Java毕设学习 基于SpringBoot + Vue 的医院管理系统 持续给大家寻找Java毕设学习项目(附源码)
基于SpringBoot + Vue的医院管理系统,涵盖医院、患者、挂号、药物、检查、病床、排班管理和数据分析等功能。开发工具为IDEA和HBuilder X,环境需配置jdk8、Node.js14、MySQL8。文末提供源码下载链接。
|
2月前
|
移动开发 前端开发 JavaScript
java家政系统成品源码的关键特点和技术应用
家政系统成品源码是已开发完成的家政服务管理软件,支持用户注册、登录、管理个人资料,家政人员信息管理,服务项目分类,订单与预约管理,支付集成,评价与反馈,地图定位等功能。适用于各种规模的家政服务公司,采用uniapp、SpringBoot、MySQL等技术栈,确保高效管理和优质用户体验。
|
2月前
|
XML JSON 监控
告别简陋:Java日志系统的最佳实践
【10月更文挑战第19天】 在Java开发中,`System.out.println()` 是最基本的输出方法,但它在实际项目中往往被认为是不专业和不足够的。本文将探讨为什么在现代Java应用中应该避免使用 `System.out.println()`,并介绍几种更先进的日志解决方案。
61 1
|
2月前
|
Java 关系型数据库 API
介绍一款Java开发的企业接口管理系统和开放平台
YesApi接口管理平台Java版,基于Spring Boot、Vue.js等技术,提供API接口的快速研发、管理、开放及收费等功能,支持多数据库、Docker部署,适用于企业级PaaS和SaaS平台的二次开发与搭建。
下一篇
DataWorks