首页 > 软件编程 > java > springboot整合rabbitmq实现订单超时取消

springboot整合rabbitmq实现订单超时取消案例分析

2025-01-26 10:29:27 作者：百万彩票中奖候选人

本文介绍了如何使用SpringBoot和RabbitMQ实现订单超时取消功能,通过配置TTL队列和死信交换机,可以管理订单的超时逻辑,实际应用中,可以通过数据库标记订单状态或手动确认机制来防止订单被错误取消

订单超时取消案例，详细请往下看~~~

1. RabbitMQ 配置类

RabbitMQConfig.java

这个类负责定义RabbitMQ的交换机、队列和绑定配置。

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    public static final String ORDER_EXCHANGE = "order-exchange";
    public static final String ORDER_QUEUE = "order-queue";
    public static final String ORDER_ROUTING_KEY = "order-routing-key";

    public static final String TTL_ORDER_QUEUE = "ttl-order-queue";
    public static final String TTL_ORDER_ROUTING_KEY = "ttl-order-routing-key";

    // 定义一个Direct类型的交换机
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }

    // 定义一个普通的队列，用于接收实际订单处理的消息
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE).build();
    }

    // 定义一个TTL（时间到期）队列，消息会在这个队列中等待TTL后转发到实际处理队列
    @Bean
    public Queue ttlOrderQueue() {
        return QueueBuilder.durable(TTL_ORDER_QUEUE)
                           .withArgument("x-dead-letter-exchange", ORDER_EXCHANGE) // 设置死信交换机
                           .withArgument("x-dead-letter-routing-key", ORDER_ROUTING_KEY) // 设置死信路由键
                           .withArgument("x-message-ttl", 60000) // 设置TTL为60秒
                           .build();
    }

    // 将实际处理队列绑定到交换机
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_ROUTING_KEY);
    }

    // 将TTL队列绑定到交换机
    @Bean
    public Binding ttlOrderBinding() {
        return BindingBuilder.bind(ttlOrderQueue()).to(orderExchange()).with(TTL_ORDER_ROUTING_KEY);
    }
}

详细解释：

交换机（Exchange）

orderExchange：定义了一个DirectExchange类型的交换机order-exchange。
Direct类型的交换机会根据路由键（routing key）精确匹配消息队列。

队列（Queue）

orderQueue：定义了一个普通的队列order-queue，这个队列用于接收和处理订单消息。
ttlOrderQueue：定义了一个TTL队列ttl-order-queue，这个队列设置了TTL（x-message-ttl）为60秒。当消息在这个队列中超过60秒未被消费，它会变成死信消息（Dead Letter），然后根据配置的死信交换机（x-dead-letter-exchange）和死信路由键（x-dead-letter-routing-key）转发到指定的队列。

绑定（Binding）

orderBinding：将order-queue队列绑定到order-exchange交换机，使用路由键order-routing-key。
ttlOrderBinding：将ttl-order-queue队列绑定到order-exchange交换机，使用路由键ttl-order-routing-key。这意味着发送到这个路由键的消息会首先进入TTL队列。

2. 订单服务

OrderService.java

这个类负责订单的创建和付款逻辑。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Service
public class OrderService {

    private static final Logger logger = LoggerFactory.getLogger(OrderService.class);

    @Autowired
    private AmqpTemplate amqpTemplate;

    // 创建订单并发送消息到TTL队列
    public void createOrder(String orderId) {
        logger.info("创建订单: {}", orderId);
        amqpTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE, RabbitMQConfig.TTL_ORDER_ROUTING_KEY, orderId);
    }

    // 支付订单
    public void payOrder(String orderId) {
        logger.info("支付订单: {}", orderId);
        // 订单支付逻辑
        // 支付成功后，需要取消TTL队列中的消息，防止订单被取消
        // 可以通过业务逻辑来实现，比如数据库状态变化
    }
}

详细解释：

createOrder方法：当创建一个订单时，会生成一个唯一的订单ID，并将其发送到order-exchange交换机，使用ttl-order-routing-key路由键。这会将消息放入TTL队列ttl-order-queue。
payOrder方法：模拟支付订单的过程。支付成功后，需要在业务逻辑中处理，确保订单不会被超时取消。这个例子没有实现具体的取消逻辑，但在实际应用中，可以通过数据库或其他机制来实现。

3. 超时监听器

OrderTimeoutListener.java

这个类负责监听超时消息并取消订单。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class OrderTimeoutListener {

    private static final Logger logger = LoggerFactory.getLogger(OrderTimeoutListener.class);

    // 监听来自order-queue队列的消息
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderTimeout(String orderId) {
        logger.info("订单超时未支付，取消订单: {}", orderId);
        // 取消订单的业务逻辑
    }
}

详细解释：

handleOrderTimeout方法：监听order-queue队列中的消息。
当TTL时间到期后，消息会被转发到这个队列，然后这个方法会被触发，处理订单超时取消的业务逻辑。

4. 主应用程序

RabbitMqOrderApplication.java

这个类是Spring Boot的主应用程序类，包含了启动逻辑和示例订单处理流程。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitMqOrderApplication implements CommandLineRunner {

    @Autowired
    private OrderService orderService;

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

    @Override
    public void run(String... args) throws Exception {
        String orderId = UUID.randomUUID().toString();
        orderService.createOrder(orderId);
        
        // 模拟延迟支付
        Thread.sleep(30000); // 30秒后支付
        orderService.payOrder(orderId);
    }
}

详细解释：

CommandLineRunner接口：实现了这个接口的run方法会在Spring Boot应用启动后立即执行。
run方法：生成一个唯一的订单ID，调用orderService.createOrder方法创建订单，并将订单消息发送到TTL队列。然后，模拟延迟30秒后调用orderService.payOrder方法支付订单。

总结

在这个示例中，我们展示了如何使用Spring Boot和RabbitMQ实现一个简单的订单超时取消功能。通过配置TTL队列和死信交换机，可以有效地管理订单的超时逻辑。

实际应用中，可以根据具体需求调整TTL时间和业务逻辑处理订单状态。

在支付成功后需要取消TTL队列中的消息，防止订单被取消，可以通过以下几种方法来实现：

方法一：使用数据库标记和业务逻辑过滤

1.数据库标记订单状态：

在订单数据库中添加一个字段来标记订单状态，例如status字段，状态值可以是NEW、PAID、CANCELLED等。
当订单支付成功后，将订单状态更新为PAID。

2.业务逻辑过滤：

在处理超时消息时，首先检查订单的状态，如果订单已经支付（状态为PAID），则忽略取消操作。

方法二：使用消息确认机制（ACK/NACK）

手动ACK消息：

配置RabbitMQ的消息监听器，使其使用手动确认（ACK）模式。
当订单支付成功时，通过业务逻辑显式地确认消息，这样RabbitMQ就不会将消息重新发送。

具体实现示例

下面我们详细介绍如何实现这两种方法。

方法一：使用数据库标记和业务逻辑过滤

1. 修改订单服务

OrderService.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    private static final Logger logger = LoggerFactory.getLogger(OrderService.class);

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Autowired
    private OrderRepository orderRepository;

    // 创建订单并发送消息到TTL队列
    public void createOrder(String orderId) {
        Order order = new Order(orderId, "NEW");
        orderRepository.save(order);

        logger.info("创建订单: {}", orderId);
        amqpTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE, RabbitMQConfig.TTL_ORDER_ROUTING_KEY, orderId);
    }

    // 支付订单
    public void payOrder(String orderId) {
        Order order = orderRepository.findById(orderId).orElse(null);
        if (order != null && "NEW".equals(order.getStatus())) {
            order.setStatus("PAID");
            orderRepository.save(order);

            logger.info("支付订单: {}", orderId);
        }
    }
}

2. 修改订单超时监听器

OrderTimeoutListener.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class OrderTimeoutListener {

    private static final Logger logger = LoggerFactory.getLogger(OrderTimeoutListener.class);

    @Autowired
    private OrderRepository orderRepository;

    // 监听来自order-queue队列的消息
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderTimeout(String orderId) {
        Order order = orderRepository.findById(orderId).orElse(null);
        if (order != null && "NEW".equals(order.getStatus())) {
            order.setStatus("CANCELLED");
            orderRepository.save(order);

            logger.info("订单超时未支付，取消订单: {}", orderId);
        } else {
            logger.info("订单已经处理: {}", orderId);
        }
    }
}

3. 订单实体和仓库

Order.java

import javax.persistence.Entity;
import javax.persistence.Id;

@Entity
public class Order {

    @Id
    private String id;
    private String status;

    // getters and setters
    public Order() {
    }

    public Order(String id, String status) {
        this.id = id;
        this.status = status;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }
}

OrderRepository.java

import org.springframework.data.repository.CrudRepository;

public interface OrderRepository extends CrudRepository<Order, String> {
}

方法二：使用消息确认机制（ACK/NACK）

1. 配置消息监听器为手动确认模式

RabbitMQConfig.java

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    // 其他配置...

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 设置手动确认
        return factory;
    }
}

2. 修改订单超时监听器以手动确认消息

OrderTimeoutListener.java

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.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

@Component
public class OrderTimeoutListener implements ChannelAwareMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(OrderTimeoutListener.class);

    @Autowired
    private OrderRepository orderRepository;

    @Override
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void onMessage(Message message, Channel channel) throws Exception {
        String orderId = new String(message.getBody());
        Order order = orderRepository.findById(orderId).orElse(null);
        if (order != null && "NEW".equals(order.getStatus())) {
            order.setStatus("CANCELLED");
            orderRepository.save(order);
            logger.info("订单超时未支付，取消订单: {}", orderId);
        } else {
            logger.info("订单已经处理: {}", orderId);
        }

        // 手动确认消息
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

通过这种方式，我们可以在订单支付成功后，通过数据库标记或手动确认机制，确保消息不会被重新发送或处理，从而防止订单被错误地取消。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

springboot整合rabbitmq实现订单超时取消案例分析

1. RabbitMQ 配置类

RabbitMQConfig.java

交换机（Exchange）

队列（Queue）

绑定（Binding）

2. 订单服务

OrderService.java

3. 超时监听器

OrderTimeoutListener.java

4. 主应用程序

RabbitMqOrderApplication.java

总结

方法一：使用数据库标记和业务逻辑过滤

方法二：使用消息确认机制（ACK/NACK）

具体实现示例

方法一：使用数据库标记和业务逻辑过滤

方法二：使用消息确认机制（ACK/NACK）

您可能感兴趣的文章: