Spring Boot + RabbitMQ 实现订单过期自动取消功能

关注

发布于: 2021 年 04 月 07 日

场景：在京东下单，订单创建成功，等待支付，一般会给 30 分钟的时间，开始倒计时。如果在这段时间内用户没有支付，则默认订单取消。

如何订单超时实现？

定时任务
redission 延时任务
rabbitmq 死信队列

本文将以 rabbitmq 死信队列展开做讲解，因为定时任务的方式，是有点问题的，原本业务系统希望 10 分钟后，如果订单未支付，就马上取消订单，并释放商品库存。但是一旦数据量大的话，就会加长获取未支付订单数据的时间，部分订单就做不到 10 分钟后取消了，可能是 15 分钟，20 分钟之类的。这样的话，库存就无法及时得到释放，也就会影响成单数。而使用 rabbitmq 死信队列，在定义业务队列时可以考虑指定一个死信交换机，并绑定一个死信队列。当消息变成死信时，该消息就会被发送到该死信队列上，再取出订单信息进行判断订单是否已支付，如未支付则讲订单状态修改为取消状态，这样也是可以达到订单超时取消的需求的。

软件准备

erlang

请参考 Win10 下安装 erlang

https://blog.csdn.net/linsongbin1/article/details/80170487

RabbitMQ

https://blog.csdn.net/linsongbin1/article/details/80170567

启动 RabbitMQ，然后添加一个用户，并给用户设置权限。

# 后台启动rabbitmq-server -detached# 添加用户rabbitmqctl add_user root 123456# 设置用户权限：rabbitmqctl set_permissions -p "/" root ".*" ".*" ".*"

复制代码

接下来创建一个 springboot 工程，并集成 RabbitMQ。

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <parent>        <groupId>org.springframework.boot</groupId>        <artifactId>spring-boot-starter-parent</artifactId>        <version>2.2.8.RELEASE</version>        <relativePath/> <!-- lookup parent from repository -->    </parent>    <groupId>com.lagou</groupId>    <artifactId>rabbitmq-work</artifactId>    <version>0.0.1-SNAPSHOT</version>    <name>rabbitmq-work</name>    <description>Demo project for Spring Boot</description>    <properties>        <java.version>1.8</java.version>    </properties>    <dependencies>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-amqp</artifactId>        </dependency>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-data-redis</artifactId>        </dependency>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency>        <dependency>            <groupId>mysql</groupId>            <artifactId>mysql-connector-java</artifactId>            <version>5.1.47</version>            <scope>runtime</scope>        </dependency>        <!--mybatis-plus依赖-->        <dependency>            <groupId>com.baomidou</groupId>            <artifactId>mybatis-plus-boot-starter</artifactId>            <version>3.3.2</version>        </dependency>
        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-test</artifactId>            <scope>test</scope>        </dependency>
    </dependencies>
    <build>        <plugins>            <plugin>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-maven-plugin</artifactId>            </plugin>        </plugins>    </build>
</project>

复制代码

接下来在 application.yml 配置文件中假如 rabbitmq 配置

spring:  application:    name: rabbit-work  # rabbitmq配置  rabbitmq:    host: localhost    virtual-host: /    username: root    password: 123456  # redis配置  redis:    timeout: 6000    host: localhost    port: 6379    database: 0  # 数据源配置  datasource:    url: jdbc:mysql://localhost:3306/order?useUnicode=true&characterEncoding=utf8&useSSL=false    driver-class-name: com.mysql.jdbc.Driver    username: root    password: 123456    hikari:      minimum-idle: 3      maximum-pool-size: 5      max-lifetime: 30000      connection-init-sql: SELECT 1# mybatis-plus配置mybatis-plus:  mapper-locations: classpath:com.lagou.*.mapper/*.xml  type-aliases-package: com.lagou.*.domain

复制代码

定义 RabbitConfig

@Configurationpublic class RabbitConfig {
    /**     * 订单队列     *     * @return {@link Queue}     */    @Bean    public Queue orderQueue() {        Map<String, Object> argments = new HashMap<>();        argments.put("x-message-ttl", 60000);        argments.put("x-dead-letter-exchange", RabbitConstants.ORDER_DLX_EXCHANGE);        argments.put("x-dead-letter-routing-key", RabbitConstants.ORDER_DLX_ROUTING_KEY);        Queue queue = new Queue(RabbitConstants.ORDER_QUEUE, true, false, false, argments);        return queue;    }
    /**     * 订单交换机     *     * @return {@link Exchange}     */    @Bean    public Exchange orderExchange() {        return new DirectExchange(RabbitConstants.ORDER_EXCHANGE, true, false, null);    }
    /**     * 订单路由键     *     * @return {@link Binding}     */    @Bean    public Binding orderRouting() {        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(RabbitConstants.ORDER_ROUTING_KEY).noargs();    }
    /**     * 订单死信队列     *     * @return {@link Queue}     */    @Bean    public Queue orderDlxQueue() {        Queue queue = new Queue(RabbitConstants.ORDER_DLX_QUEUE, true, false, false);        return queue;    }
    /**     * 订单死信交换机     *     * @return {@link Exchange}     */    @Bean    public Exchange orderDlxExchange() {        return new DirectExchange(RabbitConstants.ORDER_DLX_EXCHANGE, true, false, null);    }
    /**     * 订单死信路由键     *     * @return {@link Binding}     */    @Bean    public Binding orderDlxRouting() {        return BindingBuilder.bind(orderDlxQueue()).to(orderDlxExchange()).with(RabbitConstants.ORDER_DLX_ROUTING_KEY).noargs();    }
}

复制代码

实现消息发送

@RequestMapping(value = "/submit", produces = "application/json;charset=UTF-8")public R submit(@RequestBody OrderVo orderVo) throws UnsupportedEncodingException {  Order order = orderService.createOrder(orderVo);  System.out.println("OrderController.submit... createOrder");
  // 放入死信队列  amqpTemplate.convertAndSend(RabbitConstants.ORDER_EXCHANGE,      RabbitConstants.ORDER_ROUTING_KEY,      (order.getId() + ""));  System.out.println("OrderController.submit... sendMessage to orderExchange");  return R.ok(order);}

复制代码

消息消费者监听

@Componentpublic class OrderHandler {
    @Autowired    private OrderService orderService;
    @RabbitListener(queues = RabbitConstants.ORDER_DLX_QUEUE, ackMode = "MANUAL")    public void onMessage(Message message, Channel channel) throws IOException {        System.out.println("消息进入死信队列...");        String s = new String(message.getBody());        Order o = orderService.getById(Long.parseLong(s));        if (o != null && OrderConstants.TOPAID.getCode().equals(o.getOrderState())) {            Order order = new Order();            order.setId(o.getId());            order.setOrderState(OrderConstants.CANCEL.getCode());            order.setGmtModified(new Date());            orderService.updateById(order);        }        // 手动ack        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);    }}