文档章节

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

稳杰
 稳杰
发布于 2019/07/31 10:06
字数 2335
阅读 47
收藏 0

摘要:

本篇博文是“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://gitee.com/steadyjack/SpringBoot-SecondKill 记得Fork跟Star啊!!

2、最后,不要忘记了关注一下Debug的技术微信公众号:

© 著作权归作者所有

稳杰
粉丝 4
博文 18
码字总数 31856
作品 0
广州
高级程序员
私信 提问
加载中

评论(0)

Spring Boot(十四)RabbitMQ延迟队列

一、前言 延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单;2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度;3.过1分钟给新注册会员的用户,发送注册邮件等。 实现...

王磊的博客
2018/11/16
403
0
SpringBoot | 第三十八章:基于RabbitMQ实现消息延迟队列方案

前言 >前段时间在编写通用的消息通知服务时,由于需要实现类似通知失败时,需要延后几分钟再次进行发送,进行多次尝试后,进入定时发送机制。此机制,在原先对接银联支付时,银联的异步通知也...

oKong
2019/07/21
202
2
阿里大牛带你轻松实现RabbitMQ 延时消息

RabbitMQ 延时消息的实现(上) 我们在实际业务中有一些需要延时发送消息的场景,例如: 家里有一台智能热水器,需要在30分钟后启动 未付款的订单,15分钟后关闭 注意这里的场景是延时,不是...

Java架构
2019/01/23
0
0
实现RabbitMQ 延时消息

RabbitMQ 延时消息的实现(上) 我们在实际业务中有一些需要延时发送消息的场景,例如: 家里有一台智能热水器,需要在30分钟后启动 未付款的订单,15分钟后关闭 注意这里的场景是延时,不是...

Java架构资源分享
2019/01/23
77
0
rabbitmq 实现延迟队列的两种方式

ps: 文章里面延迟队列=延时队列 什么是延迟队列 延迟队列存储的对象肯定是对应的延时消息,所谓”延时消息”是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者...

xiaomin0322
2018/04/20
500
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周三乱弹 —— 提高不了工作效率和脸有关系

Osc乱弹歌单(2020)请戳(这里) 【今日歌曲】 @薛定谔的兄弟 :分享洛神有语创建的歌单「我喜欢的音乐」: 1 《夏令时记录(piano.ver)》- ゆめこ 手机党少年们想听歌,请使劲儿戳(这里) ...

小小编辑
今天
67
2
List的一波操作

public static void main(String[] args) { List<Entity> list = new ArrayList<>(); list.add(new Entity(1)); list.add(new Entity(2)); list.add(new Entity(3)); ......

那个猩猩很亮
今天
75
0
Spring基础

主要用于service层; 轻量级java开发框架; 各层 web层:struts,spring-MVC service层:spring dao层:hibernate,mybatis , jdbcTemplate --> spring-data Spring核心:控制反转IOC 切面编...

七宝1
今天
30
0
解决overflow+border-radius+transform圆角问题

网上还有其他版本,但是对我来说都不好使,下面是我在Chrome上的代码。overflow:hidden依然是不能正常使用,换成unset就可以,读者如果有更好的解决方案,请留言,谢谢。 <figure> <img...

hi懒喵
今天
53
0
《C语言》—— 数组

书籍使我变成了一个幸福的人,使我的生活变成轻松而舒适的诗。——高尔基 本文已经收录至我的GitHub,欢迎大家踊跃star 和 issues。 https://github.com/midou-tech/articles 点关注,不迷路!...

龙跃十二
今天
84
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部