用rocketmq发送延时消息用来取消订单（spring boot 2.3.3）

一，为什么要用延时消息来取消订单?

1,为什么要取消订单

在电商的下单过程中,需要在生成订单时扣减库存，

但有可能发生这种情况：用户下了单，临时改变主意不再支付，

则订单不能无限期的保留，因为还要把占用的库存数量释放出来，

所以通常会在用户下单后半小时（或其他时长）把未支付的订单取消不再保留。

2,取消订单的方法：

通常我们会在crond中创建一个定时运行的任务，每1分钟执行一次，

把下单时间超过半小时的取出来，检查订单状态是否还是未支付，

如果仍未支付，则修改订单状态为无效，同时把库存数量加回

这个做法的缺点是数据库繁忙时会增加数据库的压力

3,rocketmq的延时消息功能可以精准的在指定时间把消息发送到消费者，

而无需扫描数据库,

在这里我们使用延时消息来实现取消订单功能

说明：刘宏缔的架构森林是一个专注架构的博客，地址：

对应的源码可以访问这里获取：

说明：作者:刘宏缔邮箱:371125307@

二，演示项目的相关信息

1,项目地址:

2,项目功能说明

演示了用rocketmq发送延时消息

3，项目结构：如图:

三，配置文件说明

1,s/

/groupIdartifactIdspring-boot-starter-web/artifactId//groupIdartifactIdrocketmq-client//version/depency!--/groupIdartifactIdfastjson//version/depency

2,receive/

/groupIdartifactIdspring-boot-starter-web/artifactId//groupIdartifactIdrocketmq-client//version/depency!--/groupIdartifactIdfastjson//version/depency

说明：两个模块的内容相同

3,receive/

=8081

说明：两个模块同时运行时，需要把端口区分开，
s不做修改，使用默认的8080端口
receive这里指定使用8081端口

四，java代码说明

1,s/

//发送的取消订单信息publicclassOrderMsg{//用户idprivateintuserId;publicintgetUserId(){;}publicvoidsetUserId(intuserId){=userId;}//订单snprivateStringorderSn;publicStringgetOrderSn(){;}publicvoidsetOrderSn(StringorderSn){=orderSn;}}

说明：要取消的订单的消息模型,

在两个模块中一致

2,s/

publicclassRocketConstants{//nameserver,有多个时用分号隔开publicstaticfinalStringNAME_SERVER="127.0.0.1:9876";//topic的名字，应该从服务端先创建好，否则会报错publicstaticfinalStringTOPIC="laoliutest";}

rocketmq需要用到的nameserver和topic名字

在两个模块中一致

3,s/

//消息生产者类@ComponentpublicclassProducer{privateStringproducerGroup="order_producer";privateDefaultMQProducerproducer;//构造publicProducer(){//创建生产者producer=newDefaultMQProducer(producerGroup);//不开启vip通道(false);//设定(_SERVER);//();}//使producer启动publicvoidstart(){try{();}catch(MQClientExceptione){();}}//返回producerpublicDefaultMQProducergetProducer(){;}//进行关闭的方法publicvoidshutdown(){();}}

生产者类

4,s/

@RestController@RequestMapping("/home")publicclassHomeController{@AutowiredprivateProducerproducer;//初始化并发送消息@RequestMapping("/s")publicStrings()throwsException{intuserId=1;//得到订单编号:DateTimeFormatterdf_year=("yyyyMMddHHmmssSSS");LocalDateTimedate=();Stringdatestr=(df_year);//消息,指定用户id和订单编号OrderMsgmsg=newOrderMsg();(userId);(userId+"_"+datestr);StringmsgJson=(msg);//生成一个信息，标签在这里手动指定Messagemessage=newMessage(,"carttag",());//delaytime的值://messageDelayLevel=1(5);//发送信息SResultsResult=().s(message);("时间:"+()+";生产者发送一条信息，内容:{"+msgJson+"},结果:{"+sResult+"}");return"success";}}

发送消息

注意延迟时间的值5对应1m,所以消费者应该会在1分钟后才收到消息

5,receive/

@ComponentpublicclassConsumer{//消费者实体对象privateDefaultMQPushConsumerconsumer;//消费者组publicstaticfinalStringCONSUMER_GROUP="order_consumer";//构造函数用来实例化对象publicConsumer()throwsMQClientException{consumer=newDefaultMQPushConsumer(CONSUMER_GROUP);(_SERVER);//指定消费模式(_FROM_LAST_OFFSET);//指定订阅主题//指定订阅标签，*代表所有标签(,"*");//注册一个消费消息的Listener//对消息的消费在这里实现//两种回调MessageListenerConcurrently为普通监听，MessageListenerOrderly为顺序监听((MessageListenerConcurrently)(msgs,context)-{//遍历接收到的消息try{for(Messagemsg:msgs){//得到消息的bodyStringbody=newString((),"utf-8");//用json转成对象OrderMsgmsgOne=(body,);//打印用户("消息:用户id:"+());//打印订单编号("消息:订单sn:"+());//打印消息内容("时间:"+()+";消费者已接收到消息-topic={"+()+"},消息内容={"+body+"}");}}catch(UnsupportedEncodingExceptione){();_LATER;}_SUCCESS;});();("消费者启动成功=======");}}

6,其他非关键代码可查看github

五，测试效果

1,分别启动两个模块

2，访问:

返回:

success

查看s的控制台:

时间:2020-09-1714:56:53.207;生产者发送一条信息，内容:{{"orderSn":"1_202009","userId":1}},结果:{SResult[sStatus=SEND_OK,msgId=C0A803D5231F42A57993559ADFC50000,offsetMsgId=7F00000100002A9F0000000000016E7B,messageQueue=MessageQueue[topic=laoliutest,brokerName=broker-a,queueId=0],queueOffset=13]}

注意发送的时间:2020-09-1714:56:53.207

查看receive的控制台:

消息:用户id:1消息:订单sn:1_202009时间:2020-09-1714:57:53.212;消费者已接收到消息-topic={laoliutest},消息内容={{"orderSn":"1_202009","userId":1}}

注意接收到的时间:2020-09-1714:57:53.212

时长整好是60秒，和我们在代码中的设置一致

六，查看springboot版本:

._________/\\/___'_____(_)______\\\\(()\___|'_|'_||'_\/_`|\\\\\\/___)||_)|||||||(_||))))'|____|.__|_||_|_||_\__,|////=========|_|==============|___/=/_/_/_/::SpringBoot::()

很赞哦!（82）