快捷導(dǎo)航

使用RabbitMQ實(shí)現(xiàn)延時(shí)消息自動(dòng)取消的案例詳解

更新時(shí)間：2024年03月07日 10:43:55 作者：在無清風(fēng)

這篇文章主要介紹了使用RabbitMQ實(shí)現(xiàn)延時(shí)消息自動(dòng)取消的簡單案例,案例代碼包括導(dǎo)包的過程和相關(guān)配置文件，本文結(jié)合代碼給大家講解的非常詳細(xì)，需要的朋友可以參考下

一、流程圖

二、導(dǎo)包

<!--消息隊(duì)列 AMQP依賴，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

三、配置文件

#消息隊(duì)列
spring:
  rabbitmq:
    host: 192.168.88.130
    port: 5672
    virtual-host: my_vhost #使用的虛擬主機(jī)
    username: root
    password: root
    listener:
      simple:
        acknowledge-mode: manual #開啟手動(dòng)應(yīng)答

四、配置類

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitMQConfig {
    /**
     * 訂單交換機(jī)
     */
    public static final String ORDER_EXCHANGE = "order_exchange";
    /**
     * 訂單隊(duì)列
     */
    public static final String ORDER_QUEUE = "order_queue";
    /**
     * 訂單路由鍵
     */
    public static final String ORDER_ROUTING = "order_routing";
    /**
     * 死信交換機(jī)
     */
    public static final String ORDER_DEAD_EXCHANGE = "order_dead_exchange";
    /**
     * 死信隊(duì)列
     */
    public static final String ORDER_DEAD_QUEUE = "order_dead_queue";
    /**
     * 死信路由鍵
     */
    public static final String ORDER_DEAD_ROUTING = "order_dead_routing";
    /**
     * 訂單交換機(jī)
     */
    @Bean("orderExchange")
    public Exchange getOrderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    /**
     * 訂單隊(duì)列
     */
    @Bean("orderQueue")
    public Queue getOrderQueue() {
        Map<String, Object> map = new HashMap<>(3);
        map.put("x-dead-letter-exchange", ORDER_DEAD_EXCHANGE);//死信交換機(jī)
        map.put("x-dead-letter-routing-key", ORDER_DEAD_ROUTING);//死信路由鍵
        map.put("x-message-ttl", 1000 * 60 * 15);//隊(duì)列過期時(shí)間
        return QueueBuilder
                .durable(ORDER_QUEUE)
                .withArguments(map)
                .build();
    }
    /**
     * 將訂單交換機(jī)與訂單隊(duì)列綁定
     */
    @Bean
    Binding orderExchangeBindingOrder(@Qualifier("orderExchange") Exchange exchange,
                                      @Qualifier("orderQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with(ORDER_ROUTING).noargs();
    }
    /**
     * 死信交換機(jī)
     */
    @Bean("orderDeadExchange")
    public Exchange getOrderDeadExchange() {
        return new DirectExchange(ORDER_DEAD_EXCHANGE);
    }
    /**
     * 死信隊(duì)列
     */
    @Bean("orderDeadQueue")
    public Queue getOrderDeadQueue() {
        return new Queue(
                ORDER_DEAD_QUEUE,//隊(duì)列名
                true,//是否持久化
                false,//是否具有排他性，只在首次聲明時(shí)可見，不允許其他用戶訪問，連接斷開時(shí)自動(dòng)刪除
                false,//是否自動(dòng)刪除，經(jīng)歷過至少一次連接后，所有消費(fèi)者都斷開了連接，此隊(duì)列會(huì)自動(dòng)刪除
                null
        );
    }
    /**
     * 將死信交換機(jī)與死信隊(duì)列綁定
     */
    @Bean
    Binding deadExchangeBindingDeadQueue(@Qualifier("orderDeadExchange") Exchange exchange,
                                         @Qualifier("orderDeadQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with(ORDER_DEAD_ROUTING).noargs();
    }
}

五、發(fā)送消息的類

import com.sky.configuration.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
/**
 * 消息隊(duì)列發(fā)送消息
 */
@Component
public class SendRabbitMQ {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * @param orderId 15分鐘后要檢查的訂單編號(hào)
     */
    public void sendDelayOrder(Long orderId) {
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,//訂單交換機(jī)
                RabbitMQConfig.ORDER_ROUTING,//訂單路由鍵
                orderId//要取消的訂單編號(hào)
        );
    }
}

六、接收消息的類

import com.rabbitmq.client.Channel;
import com.sky.configuration.RabbitMQConfig;
import com.sky.mapper.OrderMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
/**
 * 消息隊(duì)列接收消息
 */
@Component
public class ReceiveRabbitMQ {
    @Autowired
    private OrderMapper orderMapper;
    /**
     * @param orderId 要取消的訂單的編號(hào)
     * @param msg     包含了要回復(fù)的隊(duì)列
     * @param channel 有回復(fù)功能的參數(shù)
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_DEAD_QUEUE)
    public void ReceiveDeadOrder(Long orderId, Channel channel, Message msg) throws IOException {
        orderMapper.delCancelOrder(orderId);//查詢數(shù)據(jù)庫，訂單是否付款，未付款：改為已取消
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(),//應(yīng)答的消息
                false//是否批量應(yīng)答
        );
    }
}

七、在業(yè)務(wù)代碼中注入發(fā)送類，并調(diào)用發(fā)送類的發(fā)送方法

@Service
public class OrderServiceImpl implements OrderService {
    @Autowired
    private SendRabbitMQ sendRabbitMQ;
    /**
     * 用戶下單
     */
    public void submitOrder(OrdersSubmitDTO ordersSubmitDTO) {
        sendRabbitMQ.sendDelayOrder(order.getId());//發(fā)送延遲消息到消息隊(duì)列
    }
}

到此這篇關(guān)于使用RabbitMQ實(shí)現(xiàn)延時(shí)消息自動(dòng)取消的簡單案例的文章就介紹到這了,更多相關(guān)RabbitMQ延時(shí)消息自動(dòng)取消內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: