RabbitMQ(mall学习)
延迟队列
业务场景说明
用于解决用户下单以后,订单超时如何取消订单的问题。
用户进行下单操作(会有锁定商品库存、使用优惠券、积分一系列的操作);
生成订单,获取订单的id;
获取到设置的订单超时时间(假设设置的为60分钟不支付取消订单);
按订单超时时间发送一个延迟消息给RabbitMQ,让它在订单超时后触发取消订单的操作;
如果用户没有支付,进行取消订单操作(释放锁定商品库存、返还优惠券、返回积分一系列操作)。
在pom.xml中添加相关依赖
<!--消息队列相关依赖-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--lombok依赖-->
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional>
</dependency>
修改SpringBoot配置文件
spring:rabbitmq:host: localhost # rabbitmq的连接地址port: 5672 # rabbitmq的连接端口号virtual-host: /mall # rabbitmq的虚拟hostusername: mall # rabbitmq的用户名password: mall # rabbitmq的密码publisher-confirms: true #如果对异步消息需要回调必须设置为true
添加消息队列的枚举配置类QueueEnum
用于延迟消息队列及处理取消订单消息队列的常量定义,包括交换机名称、队列名称、路由键名称。
package com.macro.mall.tiny.dto;import lombok.Getter;/*** 消息队列枚举配置* Created by macro on 2018/9/14.*/
@Getter
public enum QueueEnum {/*** 消息通知队列*/QUEUE_ORDER_CANCEL("mall.order.direct", "mall.order.cancel", "mall.order.cancel"),/*** 消息通知ttl队列*/QUEUE_TTL_ORDER_CANCEL("mall.order.direct.ttl", "mall.order.cancel.ttl", "mall.order.cancel.ttl");/*** 交换名称*/private String exchange;/*** 队列名称*/private String name;/*** 路由键*/private String routeKey;QueueEnum(String exchange, String name, String routeKey) {this.exchange = exchange;this.name = name;this.routeKey = routeKey;}
}
添加RabbitMQ的配置
用于配置交换机、队列及队列与交换机的绑定关系。
package com.macro.mall.tiny.config;import com.macro.mall.tiny.dto.QueueEnum;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 消息队列配置*/
@Configuration
public class RabbitMqConfig {/*** 订单消息实际消费队列所绑定的交换机* 此处可以选择交换机的类型 direct、fanout、topic*/@BeanDirectExchange orderDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange()).durable(true).build();}/*** 订单延迟队列队列所绑定的交换机*/@BeanDirectExchange orderTtlDirect() {return (DirectExchange) ExchangeBuilder.directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange()).durable(true).build();}/*** 订单实际消费队列*/@Beanpublic Queue orderQueue() {return new Queue(QueueEnum.QUEUE_ORDER_CANCEL.getName());}/*** 订单延迟队列(死信队列)*/@Beanpublic Queue orderTtlQueue() {return QueueBuilder.durable(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getName()).withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER_CANCEL.getExchange())//到期后转发的交换机.withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey())//到期后转发的路由键.build();}/*** 将订单队列绑定到交换机*/@BeanBinding orderBinding(DirectExchange orderDirect,Queue orderQueue){return BindingBuilder.bind(orderQueue).to(orderDirect).with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());}/*** 将订单延迟队列绑定到交换机*/@BeanBinding orderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){return BindingBuilder.bind(orderTtlQueue).to(orderTtlDirect).with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());}}
交换机及队列说明
mall.order.direct(取消订单消息队列所绑定的交换机):绑定的队列为mall.order.cancel,一旦有消息以mall.order.cancel为路由键发过来,会发送到此队列。
mall.order.direct.ttl(订单延迟消息队列所绑定的交换机):绑定的队列为mall.order.cancel.ttl,一旦有消息以mall.order.cancel.ttl为路由键发送过来,会转发到此队列,并在此队列保存一定时间,等到超时后会自动将消息发送到mall.order.cancel(取消订单消息消费队列)。
添加延迟消息的发送者CancelOrderSender
用于向订单延迟消息队列(mall.order.cancel.ttl)里发送消息。
package com.macro.mall.tiny.component;import com.macro.mall.tiny.dto.QueueEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** 取消订单消息的发出者*/
@Component
public class CancelOrderSender {private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderSender.class);@Autowiredprivate AmqpTemplate amqpTemplate;public void sendMessage(Long orderId,final long delayTimes){//给延迟队列发送消息amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//给消息设置延迟毫秒值message.getMessageProperties().setExpiration(String.valueOf(delayTimes));return message;}});LOGGER.info("send delay message orderId:{}",orderId);}
}
添加取消订单消息的接收者CancelOrderReceiver
用于从取消订单的消息队列(mall.order.cancel)里接收消息。
package com.macro.mall.tiny.component;import com.macro.mall.tiny.service.OmsPortalOrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** 取消订单消息的处理者* Created by macro on 2018/9/14.*/
@Component
@RabbitListener(queues = "mall.order.cancel")
public class CancelOrderReceiver {private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderReceiver.class);@Autowiredprivate OmsPortalOrderService portalOrderService;@RabbitHandlerpublic void handle(Long orderId){LOGGER.info("receive delay message orderId:{}",orderId);portalOrderService.cancelOrder(orderId);}
}
添加OmsPortalOrderService接口
package com.macro.mall.tiny.service;import com.macro.mall.tiny.common.api.CommonResult;
import com.macro.mall.tiny.dto.OrderParam;
import org.springframework.transaction.annotation.Transactional;/*** 前台订单管理Service* Created by macro on 2018/8/30.*/
public interface OmsPortalOrderService {/*** 根据提交信息生成订单*/@TransactionalCommonResult generateOrder(OrderParam orderParam);/*** 取消单个超时订单*/@Transactionalvoid cancelOrder(Long orderId);
}
添加OmsPortalOrderService的实现类OmsPortalOrderServiceImpl
package com.macro.mall.tiny.service.impl;import com.macro.mall.tiny.common.api.CommonResult;
import com.macro.mall.tiny.component.CancelOrderSender;
import com.macro.mall.tiny.dto.OrderParam;
import com.macro.mall.tiny.service.OmsPortalOrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;/*** 前台订单管理Service* Created by macro on 2018/8/30.*/
@Service
public class OmsPortalOrderServiceImpl implements OmsPortalOrderService {private static Logger LOGGER = LoggerFactory.getLogger(OmsPortalOrderServiceImpl.class);@Autowiredprivate CancelOrderSender cancelOrderSender;@Overridepublic CommonResult generateOrder(OrderParam orderParam) {//todo 执行一系类下单操作,具体参考mall项目LOGGER.info("process generateOrder");//下单完成后开启一个延迟消息,用于当用户没有付款时取消订单(orderId应该在下单后生成)sendDelayMessageCancelOrder(11L);return CommonResult.success(null, "下单成功");}@Overridepublic void cancelOrder(Long orderId) {//todo 执行一系类取消订单操作,具体参考mall项目LOGGER.info("process cancelOrder orderId:{}",orderId);}private void sendDelayMessageCancelOrder(Long orderId) {//获取订单超时时间,假设为60分钟long delayTimes = 30 * 1000;//发送延迟消息cancelOrderSender.sendMessage(orderId, delayTimes);}}
添加OmsPortalOrderController定义接口
package com.macro.mall.tiny.controller;import com.macro.mall.tiny.dto.OrderParam;
import com.macro.mall.tiny.service.OmsPortalOrderService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;/*** 订单管理Controller*/
@Controller
@Api(tags = "OmsPortalOrderController", description = "订单管理")
@RequestMapping("/order")
public class OmsPortalOrderController {@Autowiredprivate OmsPortalOrderService portalOrderService;@ApiOperation("根据购物车信息生成订单")@RequestMapping(value = "/generateOrder", method = RequestMethod.POST)@ResponseBodypublic Object generateOrder(@RequestBody OrderParam orderParam) {return portalOrderService.generateOrder(orderParam);}
}
Rabbitmq 插件实现延迟队列
RabbitMQ 只会检查第一个消息是否过期,如果过期则丢到死信队列,如果第一个消息的延时时长很长,而第二个消息的延时时长很短,第二个消息并不会优先得到执行。
在官网上下载 https://www.rabbitmq.com/community-plugins.html,下载
rabbitmq_delayed_message_exchange 插件,然后解压放置到 RabbitMQ 的插件目录。
进入 RabbitMQ 的安装目录下的 plgins 目录,执行下面命令让该插件生效,然后重启 RabbitMQ
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
添加RabbitMQ的配置
用于配置交换机、队列及队列与交换机的绑定关系。
@Configuration
public class DelayedQueueConfig {public static final String DELAYED_QUEUE_NAME = "delayed.queue";public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";@Beanpublic Queue delayedQueue() {return new Queue(DELAYED_QUEUE_NAME);}//自定义交换机 我们在这里定义的是一个延迟交换机@Beanpublic CustomExchange delayedExchange() {Map<String, Object> args = new HashMap<>();//自定义交换机的类型args.put("x-delayed-type", "direct");return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false,args);}@Beanpublic Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,@Qualifier("delayedExchange") CustomExchangedelayedExchange) {return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();}
}
消息生产者代码
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
@GetMapping("sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime) {rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message,correlationData ->{correlationData.getMessageProperties().setDelay(delayTime);return correlationData;});log.info(" 当 前 时 间 : {}, 发送一条延迟 {} 毫秒的信息给队列 delayed.queue:{}", newDate(),delayTime, message);}
消息消费者代码
public static final String DELAYED_QUEUE_NAME = "delayed.queue";
@RabbitListener(queues = DELAYED_QUEUE_NAME)
public void receiveDelayedQueue(Message message){String msg = new String(message.getBody());log.info("当前时间:{},收到延时队列的消息:{}", new Date().toString(), msg);}
发起请求:
http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000
RabbitMQ(mall学习)相关推荐
- rabbitMQ的学习第一篇(持续更新)
rabbitMQ的学习目标 需要带着目的去学习一项技术,在学习中深入思考感受, 1:mq的使用场景 2:几种mq什么时候选择rabbitMQ 3:MQ保存和消费消息的性能于其他MQ的对比 4:系统的q ...
- RabbitMQ深入学习指导
本文收集了一些关于RabitMQ学习参考的资料,以备学习之用.最初对RabbitMQ学习基于一次愉快的尝试,但是学习怎可不深入之?每一种MQ都解决了不同的问题,各有所长,所以深入学习和研究就是很有必要 ...
- RabbitMQ基础学习
一.RabbitMQ介绍 MQ全称为Message Queue,即消息队列, RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协 ...
- 【中间件】RabbitMQ入门学习笔记
1 .消息队列 1.1. MQ的相关概念 (1) 什么是MQ MQ(messagequeue),从字面意思上看,本质是个队列,FIFO先入先出,只不过队列中存放的内容是message而已,还是一种跨进 ...
- mall学习01-整合mybatis
1-mall学习01-概览 学习链接 http://www.macrozheng.com/#/architect/mall_arch_01?id=mysql%e6%95%b0%e6%8d%ae%e5% ...
- RabbitMQ初步学习(Mac)
1.RabbitMQ学习: 1.简介 2.安装 3.使用 3.1.创建简单列队 3.2.创建工作列队 3.3.创建订阅列队 3.4.创建路由列队 3.5.创建主题列队 3.6.事务 3.7.确认模式 ...
- 黑马RabbitMQ初级学习笔记
本文章根据b站黑马教育-rabbitMQ消息中间件视频总结所写,如有需求,请通过如下链接观看视频 2020黑马--消息中间件RabbitMQ 1.MQ的基本概念 1.1.MQ概述 MQ 全称Messa ...
- RabbitMq个人学习笔记快速掌握消息中间件
学习地址: 尚硅谷2021新版RabbitMQ教程丨快速掌握MQ消息中间件 01-入门 官网地址 https://www.rabbitmq.com/download.html 文件上传 上传到/usr ...
- rabbitmq php 学习
参考文档: http://www.cnblogs.com/phpinfo/p/4104551... http://blog.csdn.net/historyasamirror/ar... 依赖包安装 ...
最新文章
- java 轻量级 web 框架,Fast-FrameWork
- windows下配置Python环境并安装Pycharm2018
- qt creator 构建(build) 执行cmake 部署的区别?(未解决)
- java线程池_Java多线程并发:线程基本方法+线程池原理+阻塞队列原理技术分享...
- python中运行结果为false-python 两个一样的字符串用==结果为false问题的解决
- utools插件开发之监听按键的命令和状态
- 视频:如何把应用通过vmc部署到cloud foundry上
- 最适合物联网的开源数据库
- Jan 11 - Contains Duplicate II; Array; Traverse; HashMap; HashSet;
- BP神经网络用于预测
- 车路协同场景身份认证及 V2X 通信安全保障
- supplicant
- 你觉得让Android 开发所向往的高薪岗位有哪些?
- 墨尔本大学计算机科学要求,墨尔本大学计算机科学
- 计算机专业黑板报迎新,新学期迎新黑板报
- 步进电机驱动之相数、步距角、细分、拍数
- 微信小游戏保存自定义分享图到相册实例
- 盘点那些免费视频网址到底有多爽
- [分享]豪杰超级DVD播放器Ⅲ破解之菜鸟了解PE文件
- MySql练习---[SQL查询语句练习]
热门文章
- linux常用命令 java,Java工程在Linux常用命令
- Convex Hull (ACM-ICPC 2018 沈阳赛区网络预赛) 存个公式
- D. Salary Changing(二分,前缀和,贪心,瞎搞)
- Codeforces Round #633 B. Edge Weight Assignment 结论题 + dp
- 【LOJ6072】苹果树【折半搜索】【矩阵树定理】【二项式反演】
- P2567 [SCOI2010]幸运数字
- Matrix Subtraction(小米icpc邀请赛第一场)
- 牛客网 【每日一题】5月11日题目精讲 Moovie Mooving
- [指数型生成函数专练]chocolate,红色病毒问题,排列组合,字串数
- 欢乐纪中A组周六赛【2019.3.30】