天天看點
傳回

springboot內建rabbitmq 延遲隊列

源碼

  1. 引入開發包 
    <?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>springboot-integrate</artifactId>
        <groupId>springboot-integrate</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>springboot-rabbitmq</artifactId>

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

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.integrate.rabbitmq.RabbitmqApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>
  2. 将 rabbitmq 配置資訊寫入 application.yml 檔案,springBoot會建立一個 RabbitTemplate 執行個體 
    spring:
  application:
    name: springboot-rabbitmq
  rabbitmq:
    host: 52.83.239.30
    port: 5672
    username: root
    password: yunduan2019
    # 開啟發送确認
    publisher-confirms: true
    # 開啟發送失敗退回
    publisher-returns: true
    # 開啟ACK
    listener:
      direct:
        acknowledge-mode: manual
      simple:
        acknowledge-mode: manual
    template:
      mandatory: true
server:
  port: 10002
  3. 建立幾個隊列和交換機，并将隊列綁定至交換機 
    package com.integrate.rabbitmq.config;

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;

/**
 * @author 劉志強
 * @date 2020/8/18 14:07
 */
@Configuration
public class RabbitmqConfig {

    /**
     * 建立一個隊列
     * @return
     */
    @Bean(name="queue")
    public Queue queue() {
        return new Queue("queue");
    }

    @Bean(name="memberQueue")
    public Queue memberQueue() {
        return new Queue("memberQueue");
    }



    /**
     * 建立路由交換機 根據路由比對轉發消息給隊列
     * @return
     */
    @Bean(name = "exchange")
    public TopicExchange exchange() {
        return new TopicExchange("exchange");
    }


    /**
     * 配置交換機(廣播) 轉發消息給旗下所有隊列
     * @return
     */
    @Bean(name = "fanoutExchange")
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }



    /**
     * 将隊列進綁定至路由交換機并設定路由鍵
     * 交換機會将消息傳遞給 滿足路由鍵的隊列
     * @param queue
     * @param exchange
     * @return
     */
    @Bean
    Binding bindingExchangeQueue(@Qualifier("queue") Queue queue, @Qualifier("exchange") TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("exchange.queue");
    }

    @Bean
    Binding bindingExchangeMemberQueue(@Qualifier("memberQueue") Queue memberQueue, @Qualifier("exchange") TopicExchange exchange) {
        // *表示一個詞,#表示零個或多個詞
        return BindingBuilder.bind(memberQueue).to(exchange).with("exchange.*");
    }

    /**
     * 将隊列綁定至廣播交換機
     * 因為不綁定路由鍵 是以交換機會把消息傳遞給被綁定的所由隊列 廣播交換機無法設定路由鍵。因為消息會發給旗下的所有隊列
     * @param queue
     * @param fanoutExchange
     * @return
     */
    @Bean
    Binding bindingFanoutExchangeQueueTwo(@Qualifier("queue") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }
    @Bean
    Binding bindingFanoutExchangeMemberQueueTwo(@Qualifier("memberQueue") Queue memberQueue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(memberQueue).to(fanoutExchange);
    }
}
  4. 建立訂閱類，用于訂閱隊列 
    package com.integrate.rabbitmq.consumer;


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.io.IOException;

/**
 * 消息的辨別，false隻确認目前一個消息收到，true确認所有consumer獲得的消息
 * channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
 * ack傳回false，并重新回到隊列，api裡面解釋得很清楚
 * channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
 * 拒絕消息
 * channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
 * 丢棄這條消息
 * channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
 * @author 劉志強
 * @date 2020/11/13 13:45
 */
@Component
@Slf4j
public class RabbitConsumer {

    /**
     * 訂閱queue隊列 消費消息
     * @param object
     */
    @RabbitListener(queues="queue")
    public void consumerQueue(Channel channel, Object object, Message message) throws IOException {
        log.info("consumerQueue 消費來自queue隊列消息，消息内容為" + object);
        // 告訴Rabbit已消費，從隊列中删除
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    /**
     * 訂閱queue隊列 不消費消息，并将消息傳回隊列
     * @param object
     */
    @RabbitListener(queues="queue")
    public void noConsumerQueue(Channel channel, Object object, Message message) throws IOException {
        log.info("noConsumerQueue 消費來自queue隊列消息，消息内容為" + object);
        // 不消費并傳回隊列
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);

    }


    /**
     * 訂閱memberQueue隊列
     * @param object
     */
    @RabbitListener(queues="memberQueue")
    public void consumerMemberQueue(Channel channel, Object object, Message message) throws IOException {
        log.info("consumerMemberQueue 消費memberQueue隊列消息，消息内容為" + object);
        // 告訴Rabbit已消費，從隊列中删除
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

}
  5. 建立生産類 用于發送消息，并設定發送成功回調 
    package com.integrate.rabbitmq.porducer;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @author 劉志強
 * @date 2020/8/18 14:47
 */
@Component
@Slf4j
public class AckSender implements  RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 指定隊列發送
     * @param queue
     * @param content
     * @return
     */
    public CorrelationData convertAndSend(String queue, Object content) {
        //設定回調對象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //建構回調傳回的資料
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(queue,content,correlationData);
        return correlationData;
    }

    /**
     * 指定交換機 和 路由建 發送
     * @param exchange
     * @param routingKey
     * @param content
     * @return
     */
    public CorrelationData convertAndSend(String exchange, String routingKey, Object content) {
        //設定回調對象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //建構回調傳回的資料
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(exchange,routingKey, content,correlationData);
        return correlationData;
    }

    /**
     * 消息回調确認方法
     * 如果消息沒有到exchange,則confirm回調,ack=false
     * 如果消息到達exchange,則confirm回調,ack=true
     * @param
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean isSendSuccess, String s) {
        log.info("confirm--message:回調消息ID為: " + correlationData.getId());
        if (isSendSuccess) {
            log.info("消息進入交換機成功");
        } else {
            log.info("消息進入交換機失敗====" + s);
        }
    }

    /**
     * exchange到queue成功,則不回調return
     * exchange到queue失敗,則回調return(需設定mandatory=true,否則不回回調,消息就丢了)
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        StringBuilder str = new StringBuilder();
        str.append("return--message:").append(message.getBody())
                .append(",replyCode:").append(replyCode).append(",replyText:").append(replyText).append(",exchange:").append(exchange)
                .append(",routingKey:").append(routingKey);
        log.info(String.valueOf(str));
    }
}
  6. 測試 
@Autowired
    private AckSender ackSender;
    
    ackSender.convertAndSend("queue", msg);
    
    ackSender.convertAndSend("exchange", "exchange.queue", msg);

延遲消息（實作定時需求）

  1. 基于TTL(消息存活時間) 和 x-dead-letter-exchange(死信)
  2. 意思就是 消息在死亡後，會将此消息發送至死信交換機。死信交換機在發送至（x-dead-letter-routing-key）死信路由的下的死信隊列（死信交換機和死信隊列也是普通的交換機和隊列）
  3. 開始操作
    1. 建立 延遲隊列 并綁定死信交換機 及 死信路由 
      /**
    * 建立一個隊列作為延遲隊列
    * @return
    */
    @Bean(name="delayQueue")
    public Queue delayQueue() {
        Map<String, Object> args = new HashMap<>(2);
        // x-dead-letter-exchange    将隊列綁定至交換機
        args.put("x-dead-letter-exchange", "deathExchange");
        // x-dead-letter-routing-key  當消息死亡時,轉發給delayExchange交換機的路由
        args.put("x-dead-letter-routing-key", "death-route");
        return new Queue("delayQueue",true, false, false,args);
    }
    2. 建立一個交換機作為 死信交換機，延遲隊列裡死亡的消息将像此交換機發送 
      /**
     * 建立一個交換機 用于綁定死信隊列
     * @return
     */
    @Bean(name = "deathExchange")
    public TopicExchange deathExchange() {
        return new TopicExchange("deathExchange");
    }
    3. 建立一個隊列 作為死信隊列。延遲隊列裡死亡的消息将通過 交換機發送至此隊列 
      /**
     * 建立一個隊列作為死信隊列
     * @return
     */
    @Bean(name="deathQueue")
    public Queue deathQueue() {
        return new Queue("deathQueue");
    }
    4. 将死信隊列綁定至死信交換機 
      /**
     * 将死信列綁定至交換機 并設定路由健
     * @param queue
     * @param exchange
     * @return
     */
    @Bean
    Binding deathExchangeDeathQueue(@Qualifier("deathQueue") Queue queue, @Qualifier("deathExchange") TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("death-route");
    }
    5. 建立 發送消息 配置失效時間
      1. 自定義消息處理器追加 消息資訊 MessagePostProcessor 
        package com.integrate.rabbitmq.porducer;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

/**
 * @author 劉志強
 * @date 2020/11/24 16:40
 */
public class MyMessagePostProcessor implements MessagePostProcessor {

    private final Long ttl;

    public MyMessagePostProcessor(final Long ttl) {
        this.ttl = ttl;
    }

    @Override
    public Message postProcessMessage(final Message message) throws AmqpException {
        // 設定消息失效時間
        message.getMessageProperties().setExpiration(ttl.toString());
        return message;
    }
}
      2. 建立發送定時過期消息工具方法 
        public CorrelationData convertAndSendDelay(String queue, Object content, Long time) {
        //設定回調對象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //建構回調傳回的資料
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        MessagePostProcessor messagePostProcessor = new MyMessagePostProcessor(time);
        rabbitTemplate.convertAndSend(queue, content,messagePostProcessor,correlationData);
        return correlationData;
    }
    6. 訂閱死信隊列 
      /**
     * 訂閱死信隊列，當延遲隊列的消息死亡時。消息會進入死信隊列
     * @param channel
     * @param object
     * @param message
     * @throws IOException
     */
    @RabbitListener(queues="deathQueue")
    public void deathQueue(Channel channel, Object object, Message message) throws IOException {
        log.info("deathQueue 消費deathQueue隊列消息，消息内容為" + object);
        // 告訴Rabbit已消費，從隊列中删除
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
    7. 測試 
      /**
     * 像隊列發送消息并設定 TTL(過期事件)
     * @param msg
     * @return
     */
    @GetMapping("convertAndSendDelay")
    public String convertAndSendDelay(String msg,Long ttl) {
        ackSender.convertAndSendDelay("delayQueue",msg,ttl);
        return "已發送";
    }
springboot內建系列 rabbitmq springboot 延遲隊列
上一篇: Qt學習（2）——建立第一個程式
下一篇: QT學習-第一個程式

繼續閱讀