RabbitMQ延時隊列

什麼是MQ

消息隊列（Message Queue，簡稱MQ），從字面意思上看，本質是個隊列，FIFO先入先出，隻不過隊列中存放的内容是message而已。

其主要用途：不同程序Process / 線程Thread之間通信。

會産生消息隊列的原因：

不同程序（process）之間傳遞消息時，兩個程序之間耦合程度過高，改動一個程序，引發必須修改另一個程序，為了隔離這兩個程序，在兩程序間抽離出一層（一個子產品），所有兩程序之間傳遞的消息，都必須通過消息隊列來傳遞，單獨修改某一個程序，不會影響另一個；
不同程序（process）之間傳遞消息時，為了實作标準化，将消息的格式規範化了，并且，某一個程序接受的消息太多，一下子無法處理完，并且也有先後順序，必須對收到的消息進行排隊，是以誕生了事實上的消息隊列；

MQ架構非常之多，比較流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿裡開源的RocketMQ。

RabbitMQ

RabbitMQ簡介

MQ，消息隊列是應用程式和應用程式之間的通信方法
RabbitMQ是一個開源的，在AMQP基礎上完整的，可複用的企業消息系統。
支援主流的作業系統，Linux、Windows、MaxOX等。
多種開發語言支援，Java、Ptyhon、Ruby、.NET、PHP、C/C++、node.js等

安裝教程：https://blog.csdn.net/m0_37034294/article/details/82839494

SpringBoot使用RabbitMQ延時隊列

延時隊列的使用場景：

訂單業務：在電商中，使用者下單後30分鐘後未付款則取消訂單。
短信通知：使用者下單并付款後，1分鐘後發短信給使用者。

添加依賴

在 pom.xml 中添加 spring-boot-starter-amqp的依賴

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

配置application.yml

spring:
rabbitmq:

host: localhost

port: 5672

username: guest

password: guest

具體編碼實作

配置隊列

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.util.HashMap;
import java.util.Map;
 
@Configuration
@Slf4j
public class DelayRabbitConfig {
 
 
    /**
     * 延遲隊列 TTL 名稱
     */
    private static final String ORDER_DELAY_QUEUE = "user.order.delay.queue";
    /**
     * DLX，dead letter發送到的 exchange
     * 延時消息就是發送到該交換機的
     */
    public static final String ORDER_DELAY_EXCHANGE = "user.order.delay.exchange";
    /**
     * routing key 名稱
     * 具體消息發送在該 routingKey 的
     */
    public static final String ORDER_DELAY_ROUTING_KEY = "order_delay";
 
    public static final String ORDER_QUEUE_NAME = "user.order.queue";
    public static final String ORDER_EXCHANGE_NAME = "user.order.exchange";
    public static final String ORDER_ROUTING_KEY = "order";
 
    /**
     * 延遲隊列配置
     * <p>
     * 1、params.put("x-message-ttl", 5 * 1000);
     * 第一種方式是直接設定 Queue 延遲時間 但如果直接給隊列設定過期時間,這種做法不是很靈活,（當然二者是相容的,預設是時間小的優先）
     * 2、rabbitTemplate.convertAndSend(book, message -> {
     * message.getMessageProperties().setExpiration(2 * 1000 + "");
     * return message;
     * });
     * 第二種就是每次發送消息動态設定延遲時間,這樣我們可以靈活控制
     **/
    @Bean
    public Queue delayOrderQueue() {
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 聲明了隊列裡的死信轉發到的DLX名稱，
        params.put("x-dead-letter-exchange", ORDER_EXCHANGE_NAME);
        // x-dead-letter-routing-key 聲明了這些死信在轉發時攜帶的 routing-key 名稱。
        params.put("x-dead-letter-routing-key", ORDER_ROUTING_KEY);
        return new Queue(ORDER_DELAY_QUEUE, true, false, false, params);
    }
    /**
     * 需要将一個隊列綁定到交換機上，要求該消息與一個特定的路由鍵完全比對。
     * 這是一個完整的比對。如果一個隊列綁定到該交換機上要求路由鍵 “dog”，則隻有被标記為“dog”的消息才被轉發，
     * 不會轉發dog.puppy，也不會轉發dog.guard，隻會轉發dog。
     * @return DirectExchange
     */
    @Bean
    public DirectExchange orderDelayExchange() {
        return new DirectExchange(ORDER_DELAY_EXCHANGE);
    }
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayOrderQueue()).to(orderDelayExchange()).with(ORDER_DELAY_ROUTING_KEY);
    }
 
    @Bean
    public Queue orderQueue() {
        return new Queue(ORDER_QUEUE_NAME, true);
    }
    /**
     * 将路由鍵和某模式進行比對。此時隊列需要綁定要一個模式上。
     * 符号“#”比對一個或多個詞，符号“*”比對不多不少一個詞。是以“audit.#”能夠比對到“audit.irs.corporate”，但是“audit.*” 隻會比對到“audit.irs”。
     **/
    @Bean
    public TopicExchange orderTopicExchange() {
        return new TopicExchange(ORDER_EXCHANGE_NAME);
    }
 
    @Bean
    public Binding orderBinding() {
        // TODO 如果要讓延遲隊列之間有關聯,這裡的 routingKey 和 綁定的交換機很關鍵
        return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(ORDER_ROUTING_KEY);
    }
 
}

建立一個Order實體類

import lombok.Data;
 
import java.io.Serializable;
 
@Data
public class Order implements Serializable {
 
    
    private String orderId; // 訂單id
 
    private Integer orderStatus; // 訂單狀态 0：未支付，1：已支付，2：訂單已取消
 
    private String orderName; // 訂單名字
}

接收者

import com.lzc.rabbitmq.dataobject.Order;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.util.Date;
 
@Component
@Slf4j
public class DelayReceiver {
 
    @RabbitListener(queues = {DelayRabbitConfig.ORDER_QUEUE_NAME})
    public void orderDelayQueue(Order order, Message message, Channel channel) {
        log.info("###########################################");
        log.info("【orderDelayQueue 監聽的消息】 - 【消費時間】 - [{}]- 【訂單内容】 - [{}]",  new Date(), order.toString());
        if(order.getOrderStatus() == 0) {
            order.setOrderStatus(2);
            log.info("【該訂單未支付，取消訂單】" + order.toString());
        } else if(order.getOrderStatus() == 1) {
            log.info("【該訂單已完成支付】");
        } else if(order.getOrderStatus() == 2) {
            log.info("【該訂單已取消】");
        }
        log.info("###########################################");
    }
}

發送者

import com.lzc.rabbitmq.dataobject.Order;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import java.util.Date;
 
@Component
@Slf4j
public class DelaySender {
 
    @Autowired
    private AmqpTemplate amqpTemplate;
 
    public void sendDelay(Order order) {
        log.info("【訂單生成時間】" + new Date().toString() +"【1分鐘後檢查訂單是否已經支付】" + order.toString() );
        this.amqpTemplate.convertAndSend(DelayRabbitConfig.ORDER_DELAY_EXCHANGE, DelayRabbitConfig.ORDER_DELAY_ROUTING_KEY, order, message -> {
            // 如果配置了 params.put("x-message-ttl", 5 * 1000); 那麼這一句也可以省略,具體根據業務需要是聲明 Queue 的時候就指定好延遲時間還是在發送自己控制時間
            message.getMessageProperties().setExpiration(1 * 1000 * 60 + "");
            return message;
        });
    }
}

測試，通路http://localhost:8080/sendDelay，檢視日志輸出

import com.lzc.rabbitmq.config.DelaySender;
import com.lzc.rabbitmq.config.Sender;
import com.lzc.rabbitmq.dataobject.Order;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.util.concurrent.DelayQueue;
 
@RestController
public class TestController {
 
    @Autowired
    private DelaySender delaySender;
 
    @GetMapping("/sendDelay")
    public Object sendDelay() {
        Order order1 = new Order();
        order1.setOrderStatus(0);
        order1.setOrderId("123456");
        order1.setOrderName("小米6");
 
        Order order2 = new Order();
        order2.setOrderStatus(1);
        order2.setOrderId("456789");
        order2.setOrderName("小米8");
 
        delaySender.sendDelay(order1);
        delaySender.sendDelay(order2);
        return "ok";
    }
}

到延遲時間後日志輸出

【orderDelayQueue 監聽的消息】 - 【消費時間】 - [Mon Jun 18 11:56:36 CST 2018]- 【訂單内容】 - [Order(orderId=456789, orderStatus=1, orderName=小米8)]

【該訂單已完成支付】

SpringBoot使用RabbitMQ延時隊列RabbitMQ延時隊列SpringBoot使用RabbitMQ延時隊列

RabbitMQ延時隊列

什麼是MQ

RabbitMQ

SpringBoot使用RabbitMQ延時隊列

添加依賴

配置application.yml

具體編碼實作

繼續閱讀

隊列的基本用法

資料結構-隊列part 6 隊列1 定義2 隊列的順序表示——循環隊列3 隊列的鍊式存儲4 總結回顧5 球鐘問題

RabbitMQ安裝及配置前言介紹Erlang安裝RabbitMQ安裝

RabbitMQ：消費端自定義監聽

RabbitMQ：整合Spring Boot

使用阻塞隊列實作并發資料安全

線性規劃網絡流之最短增廣路算法

RabbitMQ：MessageConverter消息轉換器

rabbitmq連接配接出錯，記錄一下

RabbitMQ：交換機（fanout exchange）

第四章棧與隊列

rabbitmq 指令行操作

Rabbitmq指令行手動建立隊列

RabbitMQ指令行的基本操作

筆試面試題目：滑動視窗(二)

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？