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

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

1.订单失效问题

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

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

2.定时任务处理

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

3.延时任务处理

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

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

Java自带的DelayedQuene队列

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

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

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

RabbitMQ死信队列实现

重点介绍这种方式

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

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

【延时任务处理、订单失效】RabbitMQ死信队列实现【订单失效】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

项目结构

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

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: [email protected]
 * @createDate: 2020/12/18 8:09 上午
 * @updateUser: nxq email: [email protected]
 * @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: [email protected]
 * @createDate: 2020/12/18 8:20 上午
 * @updateUser: nxq email: [email protected]
 * @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: [email protected]
 * @createDate: 2020/12/18 8:30 上午
 * @updateUser: nxq email: [email protected]
 * @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

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

查看控制台，10秒后

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

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

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

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

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

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

转载请注明出处：https://laoniu.blog.csdn.net/article/details/111352242

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

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

1.订单失效问题

2.定时任务处理

3.延时任务处理

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

5.具体实现

5.1.环境/版本一览：

5.2.项目搭建

6.启动测试

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method