一，为什么要用延时消息来取消订单?
1,为什么要取消订单
在电商的下单过程中,需要在生成订单时扣减库存，
但有可能发生这种情况：用户下了单，临时改变主意不再支付，
则订单不能无限期的保留，因为还要把占用的库存数量释放出来，
所以通常会在用户下单后半小时（或其他时长）把未支付的订单取消不再保留。
2,取消订单的方法：
通常我们会在crond中创建一个定时运行的任务，每1分钟执行一次，
把下单时间超过半小时的取出来，检查订单状态是否还是未支付，
如果仍未支付，则修改订单状态为无效，同时把库存数量加回
这个做法的缺点是数据库繁忙时会增加数据库的压力
3,rocketmq的延时消息功能可以精准的在指定时间把消息发送到消费者，
而无需扫描数据库,
在这里我们使用延时消息来实现取消订单功能
说明：刘宏缔的架构森林是一个专注架构的博客，地址：https://www.cnblogs.com/architectforest
对应的源码可以访问这里获取： https://github.com/liuhongdi/
说明：作者:刘宏缔 邮箱: 371125307@qq.com
二，演示项目的相关信息
1,项目地址:
https://github.com/liuhongdi/mqdelay
2,项目功能说明
演示了用rocketmq发送延时消息
3，项目结构：如图:
三，配置文件说明
1,send/pom.xml
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.7.1</version> </dependency> <!--fastjson begin--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.73</version> </dependency>
2,receive/pom.xml
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.7.1</version> </dependency> <!--fastjson begin--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.73</version> </dependency>
说明：两个模块的pom.xml内容相同
3,receive/application.properties
server.port=8081
说明：两个模块同时运行时，需要把端口区分开，
send不做修改，使用默认的8080端口
receive这里指定使用8081端口
四，java代码说明
1,send/OrderMsg.java
//发送的取消订单信息 public class OrderMsg { //用户id private int userId; public int getUserId() { return this.userId; } public void setUserId(int userId) { this.userId = userId; } //订单sn private String orderSn; public String getOrderSn() { return this.orderSn; } public void setOrderSn(String orderSn) { this.orderSn = orderSn; } }
说明：要取消的订单的消息模型,
OrderMsg.java在两个模块中一致
2,send/RocketConstants.java
public class RocketConstants {//name server,有多个时用分号隔开 public static final String NAME_SERVER = "127.0.0.1:9876"; //topic的名字，应该从服务端先创建好，否则会报错 public static final String TOPIC = "laoliutest"; }
rocketmq需要用到的name server和topic名字
RocketConstants.java在两个模块中一致
3,send/Producer.java
//消息生产者类 @Component public class Producer { private String producerGroup = "order_producer"; private DefaultMQProducer producer; //构造 public Producer(){ //创建生产者 producer = new DefaultMQProducer(producerGroup); //不开启vip通道 producer.setVipChannelEnabled(false); //设定 name server producer.setNamesrvAddr(RocketConstants.NAME_SERVER); //producer.m start(); } //使producer启动 public void start(){ try { this.producer.start(); } catch (MQClientException e) { e.printStackTrace(); } } //返回producer public DefaultMQProducer getProducer(){ return this.producer; } //进行关闭的方法 public void shutdown(){ this.producer.shutdown(); } }
生产者类
4,send/HomeController.java
@RestController @RequestMapping("/home") public class HomeController { @Autowired private Producer producer; //初始化并发送消息 @RequestMapping("/send") public String send() throws Exception { int userId = 1; //得到订单编号: DateTimeFormatter df_year = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"); LocalDateTime date = LocalDateTime.now(); String datestr = date.format(df_year); //消息,指定用户id和订单编号 OrderMsg msg = new OrderMsg(); msg.setUserId(userId); msg.setOrderSn(userId+"_"+datestr); String msgJson = JSON.toJSONString(msg); //生成一个信息，标签在这里手动指定 Message message = new Message(RocketConstants.TOPIC, "carttag", msgJson.getBytes()); //delaytime的值: //messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h message.setDelayTimeLevel(5); //发送信息 SendResult sendResult = producer.getProducer().send(message); System.out.println("时间:"+ TimeUtil.getTimeNow()+";生产者发送一条信息，内容:{"+msgJson+"},结果:{"+sendResult+"}"); return "success"; } }
发送消息
注意延迟时间的值5对应1m,所以消费者应该会在1分钟后才收到消息
5,receive/Consumer.java
@Component public class Consumer { //消费者实体对象 private DefaultMQPushConsumer consumer; //消费者组 public static final String CONSUMER_GROUP = "order_consumer"; //构造函数 用来实例化对象 public Consumer() throws MQClientException { consumer = new DefaultMQPushConsumer(CONSUMER_GROUP); consumer.setNamesrvAddr(RocketConstants.NAME_SERVER); //指定消费模式 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); //指定订阅主题 //指定订阅标签，*代表所有标签 consumer.subscribe(RocketConstants.TOPIC, "*"); //注册一个消费消息的Listener //对消息的消费在这里实现 //两种回调 MessageListenerConcurrently 为普通监听，MessageListenerOrderly 为顺序监听 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { //遍历接收到的消息 try { for (Message msg : msgs) { //得到消息的body String body = new String(msg.getBody(), "utf-8"); //用json转成对象 OrderMsg msgOne = JSON.parseObject(body, OrderMsg.class); //打印用户id System.out.println("消息:用户id:"+msgOne.getUserId()); //打印订单编号 System.out.println("消息:订单sn:"+msgOne.getOrderSn()); //打印消息内容 System.out.println("时间:"+ TimeUtil.getTimeNow()+";消费者已接收到消息-topic={"+msg.getTopic()+"}, 消息内容={"+body+"}"); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); consumer.start(); System.out.println("消费者 启动成功======="); } }
6,其他非关键代码可查看github
五，测试效果
1,分别启动两个模块
2，访问:
http://127.0.0.1:8080/home/send
返回:
success
查看send的控制台:
时间:2020-09-17 14:56:53.207;生产者发送一条信息， 内容:{{"orderSn":"1_20200917145653166","userId":1}}, 结果:{SendResult [sendStatus=SEND_OK, msgId=C0A803D5231F42A57993559ADFC50000, offsetMsgId=7F00000100002A9F0000000000016E7B, messageQueue=MessageQueue [topic=laoliutest, brokerName=broker-a, queueId=0], queueOffset=13]}
注意发送的时间:2020-09-17 14:56:53.207
查看receive的控制台:
消息:用户id:1 消息:订单sn:1_20200917145653166 时间:2020-09-17 14:57:53.212; 消费者已接收到消息-topic={laoliutest}, 消息内容={{"orderSn":"1_20200917145653166","userId":1}}
注意接收到的时间:2020-09-17 14:57:53.212
时长整好是60秒，和我们在代码中的设置一致
六，查看spring boot版本:
. ____ _ __ _ _ /\ / ___'_ __ _ _(_)_ __ __ _ ( ( )\___ | '_ | '_| | '_ / _` | \/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v2.3.3.RELEASE)