【延时任务处理、订单失效】RabbitMQ死信队列实现

【订单失效】RabbitMQ死信队列实现

之前做商城遇到一个关于订单未支付超时失效的问题，总结一下

1.订单失效问题

订单失效问题比较麻烦的地方就是如何能够实时获取失效的订单。

对于这种问题一般有两种解决方案： 定时任务处理，延时任务处理

2.定时任务处理

  1. 用户下订单后先生成订单信息，然后将该订单加入到定时任务中（30分钟后执行），当到达指定时间后检查订单状态，如果未支付则标识该订单失效。
  2. 定时去轮询数据库/缓存，看订单的状态。这种方式的问题很明显，当集群部署服务器的时候需要做分布式锁进行协调，而且实时性不高，对数据库会产生压力

3.延时任务处理

当用户下订单后，将用户的订单的标识全部发送到延时队列中，30分钟后进去消费队列中被消费，消费时先检查该订单的状态，如果未支付则标识该订单失效。

有以下几种延时任务处理方式

  • Java自带的DelayedQuene队列

这是java本身提供的一种延时队列，如果项目业务复杂性不高可以考虑这种方式。它是使用jvm内存来实现的，停机会丢失数据，扩展性不强

  • 使用redis监听key的过期来实现

当用户下订单后把订单信息设置为redis的key，30分钟失效，程序编写监听redis的key失效，然后处理订单（我也尝试过这种方式）。这种方式最大的弊端就是只能监听一台redis的key失效，集群下将无法实现，也有人监听集群下的每个redis节点的key，但我认为这样做很不合适。如果项目业务复杂性不高，redis单机部署，就可以考虑这种方式

  • RabbitMQ死信队列实现

重点介绍这种方式

4.RabbitMQ死信队列实现监听订单失效

AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。

  • Time To Live(TTL)

RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间，如果超时(两者同时设置以最先到期的时间为准)，则消息变为dead letter(死信)

A: 通过队列属性设置，队列中所有消息都有相同的过期时间。 
B: 对消息进行单独设置，每条消息TTL可以不同。

  • Dead Letter Exchanges（DLX）

RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由转发到指定的队列。
x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送

下面来做一个例子来实现订单失效，为了效果明显，我们把订单的失效时间设置为10秒 （java实现）

5.具体实现

5.1.环境/版本一览：

  • 开发工具：Intellij IDEA 2020.2.3
  • springboot：2.4.1
  • jdk：1.8.0_211
  • maven: 3.6.3
  • RabbitMQ：3.8.9

docker安装rabbitmq可以查看我这一篇博客  docker 安装rabbitMQ ：https://laoniu.blog.csdn.net/article/details/110090272

项目结构

5.2.项目搭建

最下方有完整代码

  • 新建项目，配置pom.xml加入依赖
<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
  • 配置RabbitMQ的队列和路由信息  创建类 RabbitMQConfiuration
package com.niu.springbootrabbitmqdelayqueue.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


import java.util.HashMap;
import java.util.Map;

/**
 * @description: rabbitMQ配置信息
 * @author: nxq email: niuxiangqian163@163.com
 * @createDate: 2020/12/18 8:09 上午
 * @updateUser: nxq email: niuxiangqian163@163.com
 * @updateDate: 2020/12/18 8:09 上午
 * @updateRemark:
 * @version: 1.0
 **/
@Configuration
public class RabbitMQConfiguration {
    //队列名称
   public   final static String orderQueue = "order_queue";

    //交换机名称
    public  final static String orderExchange = "order_exchange";

    // routingKey
    public  final static String routingKeyOrder = "routing_key_order";

    //死信消息队列名称
    public  final static String dealQueueOrder = "deal_queue_order";

    //死信交换机名称
    public  final static String dealExchangeOrder = "deal_exchange_order";

    //死信 routingKey
    public final static String deadRoutingKeyOrder = "dead_routing_key_order";

    //死信队列 交换机标识符
    public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";

    //死信队列交换机绑定键标识符
    public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Bean
    public Queue orderQueue() {
        // 将普通队列绑定到死信队列交换机上
        Map<String, Object> args = new HashMap<>(2);
        //args.put("x-message-ttl", 5 * 1000);//直接设置 Queue 延迟时间 但如果直接给队列设置过期时间,这种做法不是很灵活
        //这里采用发送消息动态设置延迟时间,这样我们可以灵活控制
        args.put(DEAD_LETTER_QUEUE_KEY, dealExchangeOrder);
        args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKeyOrder);
        return new Queue(RabbitMQConfiguration.orderQueue, true, false, false, args);
    }

    //声明一个direct类型的交换机
    @Bean
    DirectExchange orderExchange() {
        return new DirectExchange(RabbitMQConfiguration.orderExchange);
    }

    //绑定Queue队列到交换机,并且指定routingKey
    @Bean
    Binding bindingDirectExchangeDemo5(   ) {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(routingKeyOrder);
    }

    //创建配置死信队列
    @Bean
    public Queue deadQueueOrder() {
        Queue queue = new Queue(dealQueueOrder, true);
        return queue;
    }

    //创建死信交换机
    @Bean
    public DirectExchange deadExchangeOrder() {
        return new DirectExchange(dealExchangeOrder);
    }

    //死信队列与死信交换机绑定
    @Bean
    public Binding bindingDeadExchange() {
        return BindingBuilder.bind(deadQueueOrder()).to(deadExchangeOrder()).with(deadRoutingKeyOrder);
    }



}
  • 配置消息生产者
package com.niu.springbootrabbitmqdelayqueue.controller;

import com.niu.springbootrabbitmqdelayqueue.config.RabbitMQConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

/**
 * @description: 订单的控制器
 * @author: nxq email: niuxiangqian163@163.com
 * @createDate: 2020/12/18 8:20 上午
 * @updateUser: nxq email: niuxiangqian163@163.com
 * @updateDate: 2020/12/18 8:20 上午
 * @updateRemark:
 * @version: 1.0
 **/
@RestController
@RequestMapping("/order")
public class OrderController {
    private static final Logger logger =  LoggerFactory.getLogger(OrderController.class);
    @Autowired
    private AmqpTemplate rabbitTemplate;

    /**
     * 模拟提交订单
     * @author nxq
     * @return java.lang.Object
     */
    @GetMapping("")
    public Object submit(){
        String orderId = UUID.randomUUID().toString();
        logger.info("submit order {}", orderId);
        this.rabbitTemplate.convertAndSend(
                RabbitMQConfiguration.orderExchange, //发送至订单交换机
                RabbitMQConfiguration.routingKeyOrder, //订单定routingKey
                orderId //订单号   这里可以传对象 比如直接传订单对象
                , message -> {
            // 如果配置了 params.put("x-message-ttl", 5 * 1000);
            // 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
            message.getMessageProperties().setExpiration(1000 * 10 + "");
            return message;
        });
        
        return "{'orderId':'"+orderId+"'}";
    }
}
  • 配置消费者 
package com.niu.springbootrabbitmqdelayqueue.listener;

import com.niu.springbootrabbitmqdelayqueue.config.RabbitMQConfiguration;
import com.niu.springbootrabbitmqdelayqueue.controller.OrderController;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.Map;
import com.rabbitmq.client.Channel;

/**
 * @description: 订单失效监听器
 * @author: nxq email: niuxiangqian163@163.com
 * @createDate: 2020/12/18 8:30 上午
 * @updateUser: nxq email: niuxiangqian163@163.com
 * @updateDate: 2020/12/18 8:30 上午
 * @updateRemark:
 * @version: 1.0
 **/
@Component
public class OrderFailureListener {
    private static final Logger logger =  LoggerFactory.getLogger(OrderFailureListener.class);
    @RabbitListener(
            queues = RabbitMQConfiguration.dealQueueOrder //设置订单失效的队列
    )
    public void process(String order, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {

        logger.info("【订单号】 - [{}]",  order);
        // 判断订单是否已经支付，如果支付则；否则，取消订单（逻辑代码省略)

        // 手动ack
//        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        // 手动签收
//        channel.basicAck(deliveryTag, false);
        System.out.println("执行结束....");

    }
}
  • 配置文件 application.yml 加入rabbitmq的相关配置
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin
  • 配置启动类
package com.niu.springbootrabbitmqdelayqueue;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringbootRabbitmqDelayQueueApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringbootRabbitmqDelayQueueApplication.class, args);
    }

}

6.启动测试

启动成功后我们访问一下配置的接口模拟提交订单 http://localhost:8080/order

查看控制台，10秒后

芜湖～起飞?️，大功告成，可以说实时性已经非常高了，可以试着多提交几次订单

打开rabbitMQ的控制台看一下，发现多出来两个队列

使用这种方式不止可以做订单失效，比如说优惠券过期啊等等延时失效问题。可以集群部署rabbitmq，开启消息确认机制。

这种实现方法的基本使用到此就讲完了，完整代码已经推送至github ：https://github.com/1603565290m/springboot-rabbitmq-delay-queue

转载请注明出处：https://refblogs.com/article/82

正文到此结束
所属分类：后端技术
相关文章
热门推荐
评论插件初始化中...
Loading...
标签云
自动化 加密 插入 代理模式 异步通信 jQuery 关联设计 前端开发 方法 设计原则 Caffeine 提高效率 打包 ThreadLocal SQL查询性能 管理和监控 结构体 面向对象设计原则 击穿 生僻字 Onion洋葱模型 缓存淘汰机制 创建型设计模式 ENTRYPOINT 面试 穿透 集合 C++ Git SpringBoot 网络通信 rabbitmq 小程序 Web服务器 数据持久化
本文目录