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

2022-01-24 21:53:42

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

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)