spring boot集成RabbitMQ實現訂單超時取消完整版

    最近公司做項目，涉及到下訂單的功能，項目不大，用的人也不多，其實可以不用引入rabbit mq的，但本着閑着也是閑着的態度，即使項目規模不大咱也專業點。其實之前做過類似需求的功能，當時的實現方式是每個一分鍾查詢一次數據庫，判斷當前記錄的下單時間是否超時了，然后更改訂單狀態，是不是不太professional😄。閑扯就到這里了，下面開始正文。

關於rabbit mq的安裝在這里就不說了，如果有需要請給我留言，可以單獨輔導😄。

1、引入rabbit mq依賴：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

 

spring:
  #整合rabbitmq
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

 

 2、rabbit mq配置類編寫

@Configuration
public class RabbitMqConfig {

    // 支付超時延時交換機
    public static final String Delay_Exchange_Name = "delay.exchange";

    // 超時訂單關閉隊列
    public static final String Timeout_Trade_Queue_Name = "close_trade";


    @Bean
    public Queue delayPayQueue() {
        return new Queue(RabbitMqConfig.Timeout_Trade_Queue_Name, true);
    }


    // 定義廣播模式的延時交換機 無需綁定路由
    @Bean
    FanoutExchange delayExchange(){
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type", "direct");
        FanoutExchange topicExchange = new FanoutExchange(RabbitMqConfig.Delay_Exchange_Name, true, false, args);
        topicExchange.setDelayed(true);
        return topicExchange;
    }

    // 綁定延時隊列與交換機
    @Bean
    public Binding delayPayBind() {
        return BindingBuilder.bind(delayPayQueue()).to(delayExchange());
    }

    // 定義消息轉換器
    @Bean
    Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 定義消息模板用於發布消息，並且設置其消息轉換器
    @Bean
    RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }
    @Bean
    RabbitAdmin rabbitAdmin(final ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

}

這塊注意一下，必須要安裝rabbitmq_delayed_message_exchange插件，下載地址：點我。不然啟動的時候會報錯。

 

3、實體類代碼：

import java.io.Serializable;
import java.math.BigDecimal;

public class OrderMasterDo implements Serializable {

    private String orderId;
    private BigDecimal amt;
    private Integer payStatus;

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public BigDecimal getAmt() {
        return amt;
    }

    public void setAmt(BigDecimal amt) {
        this.amt = amt;
    }

    public Integer getPayStatus() {
        return payStatus;
    }

    public void setPayStatus(Integer payStatus) {
        this.payStatus = payStatus;
    }
}

 

4、消息生產者代碼：

@RestController
@RequestMapping("/basic")
public class BasicInfoController {

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/test")
    public String createOrderTest() {

        // 創建訂單
        OrderMasterDo orderMaster = new OrderMasterDo();
        //未支付
        orderMaster.setPayStatus(0);
        orderMaster.setOrderId("001");
        orderMaster.setAmt(new BigDecimal(1000));

        // 發送訂單到消息隊列
        rabbitTemplate.convertAndSend(RabbitMqConfig.Delay_Exchange_Name, "", orderMaster, message ->{
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            message.getMessageProperties().setDelay(10*1000);   // 毫秒為單位，指定此消息的延時時長
            return message;
        });

        System.out.println("創建訂單成功");

        return "創建訂單成功";
    }
}

 

5、消費者代碼：

@Component
public class OrderReceiver {

    @RabbitListener(queues = RabbitMqConfig.Timeout_Trade_Queue_Name)
    public void process(OrderMasterDo orderMasterDo, Message message, Channel channel) throws IOException{
        try {
            System.out.println("開始執行訂單[{}]的支付超時訂單關閉......");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("超時訂單處理完畢");
        } catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}

 

6、測試效果：

 可以看到，生產者生產的消息在10s之后被消費。

 

總結：

以上就實現了基本的功能需求，但還不完美，因為沒有考慮到消息丟失、重復等問題，關於這塊下一篇會說到。

 

參考資料：

https://segmentfault.com/a/1190000016072908