springboot整合rabbitmq實現(xiàn)訂單超時取消案例分析

更新時間：2025年01月26日 10:29:52 作者：百萬彩票中獎候選人

本文介紹了如何使用SpringBoot和RabbitMQ實現(xiàn)訂單超時取消功能,通過配置TTL隊列和死信交換機,可以管理訂單的超時邏輯,實際應(yīng)用中,可以通過數(shù)據(jù)庫標(biāo)記訂單狀態(tài)或手動確認(rèn)機制來防止訂單被錯誤取消

1. RabbitMQ 配置類

RabbitMQConfig.java

這個類負(fù)責(zé)定義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后轉(zhuǎn)發(fā)到實際處理隊列
    @Bean
    public Queue ttlOrderQueue() {
        return QueueBuilder.durable(TTL_ORDER_QUEUE)
                           .withArgument("x-dead-letter-exchange", ORDER_EXCHANGE) // 設(shè)置死信交換機
                           .withArgument("x-dead-letter-routing-key", ORDER_ROUTING_KEY) // 設(shè)置死信路由鍵
                           .withArgument("x-message-ttl", 60000) // 設(shè)置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);
    }
}

詳細(xì)解釋：

交換機（Exchange）

orderExchange：定義了一個DirectExchange類型的交換機order-exchange。
Direct類型的交換機會根據(jù)路由鍵（routing key）精確匹配消息隊列。

隊列（Queue）

orderQueue：定義了一個普通的隊列order-queue，這個隊列用于接收和處理訂單消息。
ttlOrderQueue：定義了一個TTL隊列ttl-order-queue，這個隊列設(shè)置了TTL（x-message-ttl）為60秒。當(dāng)消息在這個隊列中超過60秒未被消費，它會變成死信消息（Dead Letter），然后根據(jù)配置的死信交換機（x-dead-letter-exchange）和死信路由鍵（x-dead-letter-routing-key）轉(zhuǎn)發(fā)到指定的隊列。

綁定（Binding）

orderBinding：將order-queue隊列綁定到order-exchange交換機，使用路由鍵order-routing-key。
ttlOrderBinding：將ttl-order-queue隊列綁定到order-exchange交換機，使用路由鍵ttl-order-routing-key。這意味著發(fā)送到這個路由鍵的消息會首先進(jìn)入TTL隊列。

2. 訂單服務(wù)

OrderService.java

這個類負(fù)責(zé)訂單的創(chuàng)建和付款邏輯。

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;

    // 創(chuàng)建訂單并發(fā)送消息到TTL隊列
    public void createOrder(String orderId) {
        logger.info("創(chuàng)建訂單: {}", orderId);
        amqpTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE, RabbitMQConfig.TTL_ORDER_ROUTING_KEY, orderId);
    }

    // 支付訂單
    public void payOrder(String orderId) {
        logger.info("支付訂單: {}", orderId);
        // 訂單支付邏輯
        // 支付成功后，需要取消TTL隊列中的消息，防止訂單被取消
        // 可以通過業(yè)務(wù)邏輯來實現(xiàn)，比如數(shù)據(jù)庫狀態(tài)變化
    }
}

詳細(xì)解釋：

createOrder方法：當(dāng)創(chuàng)建一個訂單時，會生成一個唯一的訂單ID，并將其發(fā)送到order-exchange交換機，使用ttl-order-routing-key路由鍵。這會將消息放入TTL隊列ttl-order-queue。
payOrder方法：模擬支付訂單的過程。支付成功后，需要在業(yè)務(wù)邏輯中處理，確保訂單不會被超時取消。這個例子沒有實現(xiàn)具體的取消邏輯，但在實際應(yīng)用中，可以通過數(shù)據(jù)庫或其他機制來實現(xiàn)。

3. 超時監(jiān)聽器

OrderTimeoutListener.java

這個類負(fù)責(zé)監(jiān)聽超時消息并取消訂單。

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);

    // 監(jiān)聽來自order-queue隊列的消息
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderTimeout(String orderId) {
        logger.info("訂單超時未支付，取消訂單: {}", orderId);
        // 取消訂單的業(yè)務(wù)邏輯
    }
}

詳細(xì)解釋：

handleOrderTimeout方法：監(jiān)聽order-queue隊列中的消息。
當(dāng)TTL時間到期后，消息會被轉(zhuǎn)發(fā)到這個隊列，然后這個方法會被觸發(fā)，處理訂單超時取消的業(yè)務(wù)邏輯。

4. 主應(yīng)用程序

RabbitMqOrderApplication.java

這個類是Spring Boot的主應(yīng)用程序類，包含了啟動邏輯和示例訂單處理流程。

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);
    }
}

詳細(xì)解釋：

CommandLineRunner接口：實現(xiàn)了這個接口的run方法會在Spring Boot應(yīng)用啟動后立即執(zhí)行。
run方法：生成一個唯一的訂單ID，調(diào)用orderService.createOrder方法創(chuàng)建訂單，并將訂單消息發(fā)送到TTL隊列。然后，模擬延遲30秒后調(diào)用orderService.payOrder方法支付訂單。

總結(jié)

在這個示例中，我們展示了如何使用Spring Boot和RabbitMQ實現(xiàn)一個簡單的訂單超時取消功能。通過配置TTL隊列和死信交換機，可以有效地管理訂單的超時邏輯。

實際應(yīng)用中，可以根據(jù)具體需求調(diào)整TTL時間和業(yè)務(wù)邏輯處理訂單狀態(tài)。

在支付成功后需要取消TTL隊列中的消息，防止訂單被取消，可以通過以下幾種方法來實現(xiàn)：

方法一：使用數(shù)據(jù)庫標(biāo)記和業(yè)務(wù)邏輯過濾

1.數(shù)據(jù)庫標(biāo)記訂單狀態(tài)：

在訂單數(shù)據(jù)庫中添加一個字段來標(biāo)記訂單狀態(tài)，例如status字段，狀態(tài)值可以是NEW、PAID、CANCELLED等。
當(dāng)訂單支付成功后，將訂單狀態(tài)更新為PAID。

2.業(yè)務(wù)邏輯過濾：

在處理超時消息時，首先檢查訂單的狀態(tài)，如果訂單已經(jīng)支付（狀態(tài)為PAID），則忽略取消操作。

方法二：使用消息確認(rèn)機制（ACK/NACK）

手動ACK消息：

配置RabbitMQ的消息監(jiān)聽器，使其使用手動確認(rèn)（ACK）模式。
當(dāng)訂單支付成功時，通過業(yè)務(wù)邏輯顯式地確認(rèn)消息，這樣RabbitMQ就不會將消息重新發(fā)送。

具體實現(xiàn)示例

下面我們詳細(xì)介紹如何實現(xiàn)這兩種方法。

方法一：使用數(shù)據(jù)庫標(biāo)記和業(yè)務(wù)邏輯過濾

1. 修改訂單服務(wù)

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;

    // 創(chuàng)建訂單并發(fā)送消息到TTL隊列
    public void createOrder(String orderId) {
        Order order = new Order(orderId, "NEW");
        orderRepository.save(order);

        logger.info("創(chuàng)建訂單: {}", 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. 修改訂單超時監(jiān)聽器

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;

    // 監(jiān)聽來自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("訂單已經(jīng)處理: {}", 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> {
}

方法二：使用消息確認(rèn)機制（ACK/NACK）

1. 配置消息監(jiān)聽器為手動確認(rèn)模式

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); // 設(shè)置手動確認(rèn)
        return factory;
    }
}

2. 修改訂單超時監(jiān)聽器以手動確認(rèn)消息

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("訂單已經(jīng)處理: {}", orderId);
        }

        // 手動確認(rèn)消息
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

通過這種方式，我們可以在訂單支付成功后，通過數(shù)據(jù)庫標(biāo)記或手動確認(rèn)機制，確保消息不會被重新發(fā)送或處理，從而防止訂單被錯誤地取消。

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: