SpringBoot集成RabbitMQ（死信隊列）

honhon 發布于2019-08-16 17:04 / 3097人閱讀

摘要：介紹死信隊列沒有被及時消費的消息存放的隊列，消息沒有被及時消費有以下幾點原因有消息被拒絕并且隊列達到最大長度消息過期場景小時進入初始隊列，等待分鐘后進入分鐘隊列消息等待分鐘后進入執行隊列執行失敗后重新回到分鐘隊列失敗次后，消息進入小時隊列消

介紹

死信隊列：沒有被及時消費的消息存放的隊列，消息沒有被及時消費有以下幾點原因：
1.有消息被拒絕（basic.reject/ basic.nack）并且requeue=false
2.隊列達到最大長度
3.消息TTL過期

場景

1.小時進入初始隊列，等待30分鐘后進入5分鐘隊列
2.消息等待5分鐘后進入執行隊列
3.執行失敗后重新回到5分鐘隊列
4.失敗5次后，消息進入2小時隊列
5.消息等待2小時進入執行隊列
6.失敗5次后，將消息丟棄或做其他處理

使用

安裝MQ

使用docker方式安裝,選擇帶mangement的版本

docker pull rabbitmq:management
docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management

訪問 localhost: 15672,默認賬號密碼guest/guest

項目配置

（1）創建springboot項目
（2）在application.properties配置文件中配置mq連接信息

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

（3）隊列配置

package com.df.ps.mq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowire;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class MqConfig {

    //time
    @Value("${spring.df.buffered.min:120}")
    private int springdfBufferedTime;

    @Value("${spring.df.high-buffered.min:5}")
    private int springdfHighBufferedTime;

    @Value("${spring.df.low-buffered.min:120}")
    private int springdfLowBufferedTime;

    // 30min Buffered Queue
    @Value("${spring.df.queue:spring-df-buffered-queue}")
    private String springdfBufferedQueue;

    @Value("${spring.df.topic:spring-df-buffered-topic}")
    private String springdfBufferedTopic;

    @Value("${spring.df.route:spring-df-buffered-route}")
    private String springdfBufferedRouteKey;

    // 5M Buffered Queue
    @Value("${spring.df.high-buffered.queue:spring-df-high-buffered-queue}")
    private String springdfHighBufferedQueue;

    @Value("${spring.df.high-buffered.topic:spring-df-high-buffered-topic}")
    private String springdfHighBufferedTopic;

    @Value("${spring.df.high-buffered.route:spring-df-high-buffered-route}")
    private String springdfHighBufferedRouteKey;

    // High Queue
    @Value("${spring.df.high.queue:spring-df-high-queue}")
    private String springdfHighQueue;

    @Value("${spring.df.high.topic:spring-df-high-topic}")
    private String springdfHighTopic;

    @Value("${spring.df.high.route:spring-df-high-route}")
    private String springdfHighRouteKey;

    // 2H Low Buffered Queue
    @Value("${spring.df.low-buffered.queue:spring-df-low-buffered-queue}")
    private String springdfLowBufferedQueue;

    @Value("${spring.df.low-buffered.topic:spring-df-low-buffered-topic}")
    private String springdfLowBufferedTopic;

    @Value("${spring.df.low-buffered.route:spring-df-low-buffered-route}")
    private String springdfLowBufferedRouteKey;

    // Low Queue
    @Value("${spring.df.low.queue:spring-df-low-queue}")
    private String springdfLowQueue;

    @Value("${spring.df.low.topic:spring-df-low-topic}")
    private String springdfLowTopic;

    @Value("${spring.df.low.route:spring-df-low-route}")
    private String springdfLowRouteKey;


    @Bean(autowire = Autowire.BY_NAME, value = "springdfBufferedQueue")
    Queue springdfBufferedQueue() {
        int bufferedTime = 1000 * 60 * springdfBufferedTime;
        return createBufferedQueue(springdfBufferedQueue, springdfHighBufferedTopic, springdfHighBufferedRouteKey, bufferedTime);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfHighBufferedQueue")
    Queue springdfHighBufferedQueue() {
        int highBufferedTime = 1000 * 60 * springdfHighBufferedTime;
        return createBufferedQueue(springdfHighBufferedQueue, springdfHighTopic, springdfHighRouteKey, highBufferedTime);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfHighQueue")
    Queue springdfHighQueue() {
        return new Queue(springdfHighQueue, true);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfLowBufferedQueue")
    Queue springdfLowBufferedQueue() {
        int lowBufferedTime = 1000 * 60 * springdfLowBufferedTime;
        return createBufferedQueue(springdfLowBufferedQueue, springdfLowTopic, springdfLowRouteKey, lowBufferedTime);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfLowQueue")
    Queue springdfLowQueue() {
        return new Queue(springdfLowQueue, true);
    }


    @Bean(autowire = Autowire.BY_NAME, value = "springdfBufferedTopic")
    TopicExchange springdfBufferedTopic() {
        return new TopicExchange(springdfBufferedTopic);
    }

    @Bean
    Binding springBuffereddf(Queue springdfBufferedQueue, TopicExchange springdfBufferedTopic) {
        return BindingBuilder.bind(springdfBufferedQueue).to(springdfBufferedTopic).with(springdfBufferedRouteKey);
    }


    @Bean(autowire = Autowire.BY_NAME, value = "springdfHighBufferedTopic")
    TopicExchange springdfHighBufferedTopic() {
        return new TopicExchange(springdfHighBufferedTopic);
    }

    @Bean
    Binding springHighBuffereddf(Queue springdfHighBufferedQueue, TopicExchange springdfHighBufferedTopic) {
        return BindingBuilder.bind(springdfHighBufferedQueue).to(springdfHighBufferedTopic).with(springdfHighBufferedRouteKey);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfHighTopic")
    TopicExchange springdfHighTopic() {
        return new TopicExchange(springdfHighTopic);
    }

    @Bean
    Binding springHighdf(Queue springdfHighQueue, TopicExchange springdfHighTopic) {
        return BindingBuilder.bind(springdfHighQueue).to(springdfHighTopic).with(springdfHighRouteKey);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfLowBufferedTopic")
    TopicExchange springdfLowBufferedTopic() {
        return new TopicExchange(springdfLowBufferedTopic);
    }

    @Bean
    Binding springLowBuffereddf(Queue springdfLowBufferedQueue, TopicExchange springdfLowBufferedTopic) {
        return BindingBuilder.bind(springdfLowBufferedQueue).to(springdfLowBufferedTopic).with(springdfLowBufferedRouteKey);
    }

    @Bean(autowire = Autowire.BY_NAME, value = "springdfLowTopic")
    TopicExchange springdfLowTopic() {
        return new TopicExchange(springdfLowTopic);
    }

    @Bean
    Binding springLowdf(Queue springdfLowQueue, TopicExchange springdfLowTopic) {
        return BindingBuilder.bind(springdfLowQueue).to(springdfLowTopic).with(springdfLowRouteKey);
    }


    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                             MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(springdfHighQueue, springdfLowQueue);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(IntegrationReceiver receiver) {


        MessageListenerAdapter adapter = new MessageListenerAdapter(receiver);
        adapter.setDefaultListenerMethod("receive");
        Map queueOrTagToMethodName = new HashMap<>();
        queueOrTagToMethodName.put(springdfHighQueue, "springdfHighReceive");
        queueOrTagToMethodName.put(springdfLowQueue, "springdfLowReceive");
        adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
        return adapter;

    }


    private Queue createBufferedQueue(String queueName, String topic, String routeKey, int bufferedTime) {
        Map args = new HashMap<>();
        args.put("x-dead-letter-exchange", topic);
        args.put("x-dead-letter-routing-key", routeKey);
        args.put("x-message-ttl", bufferedTime);
        // 是否持久化
        boolean durable = true;
        // 僅創建者可以使用的私有隊列，斷開后自動刪除
        boolean exclusive = false;
        // 當所有消費客戶端連接斷開后，是否自動刪除隊列
        boolean autoDelete = false;

        return new Queue(queueName, durable, exclusive, autoDelete, args);
    }
}

消費者配置

package com.df.ps.mq;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.Map;

public class MqReceiver {

    private static Logger logger = LoggerFactory.getLogger(MqReceiver.class);

    @Value("${high-retry:5}")
    private int highRetry;

    @Value("${low-retry:5}")
    private int lowRetry;

    @Value("${spring.df.high-buffered.topic:spring-df-high-buffered-topic}")
    private String springdfHighBufferedTopic;

    @Value("${spring.df.high-buffered.route:spring-df-high-buffered-route}")
    private String springdfHighBufferedRouteKey;

    @Value("${spring.df.low-buffered.topic:spring-df-low-buffered-topic}")
    private String springdfLowBufferedTopic;

    @Value("${spring.df.low-buffered.route:spring-df-low-buffered-route}")
    private String springdfLowBufferedRouteKey;

    private final RabbitTemplate rabbitTemplate;
    @Autowired
    public MqReceiver(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void receive(Object message) {
        if (logger.isInfoEnabled()) {
            logger.info("default receiver: " + message);
        }
    }

    /**
     * 消息從初始隊列進入5分鐘的高速緩沖隊列
     * @param message
     */
    public void highReceiver(Object message){
        ObjectMapper mapper = new ObjectMapper();
        Map msg = mapper.convertValue(message, Map.class);

        try{
            logger.info("這里做消息處理...");
        }catch (Exception e){
            int times = msg.get("times") == null ? 0 : (int) msg.get("times");
            if (times < highRetry) {
                msg.put("times", times + 1);
                rabbitTemplate.convertAndSend(springdfHighBufferedTopic,springdfHighBufferedRouteKey,message);
            } else {
                msg.put("times", 0);
                rabbitTemplate.convertAndSend(springdfLowBufferedTopic,springdfLowBufferedRouteKey,message);
            }
        }
    }

    /**
     * 消息從5分鐘緩沖隊列進入2小時緩沖隊列
     * @param message
     */
    public void lowReceiver(Object message){
        ObjectMapper mapper = new ObjectMapper();
        Map msg = mapper.convertValue(message, Map.class);
        
        try {
            logger.info("這里做消息處理...");
        }catch (Exception e){
            int times = msg.get("times") == null ? 0 : (int) msg.get("times");
            if (times < lowRetry) {
                rabbitTemplate.convertAndSend(springdfLowBufferedTopic,springdfLowBufferedRouteKey,message);
            }else{
                logger.info("消息無法被消費...");
            }
        } 
    }
}

GPU云服務器云服務器死信隊列取出消息死信隊列重新消費死信 RabbitMQ

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/74366.html

SpringBoot+RabbitMq實現延時消息隊列

背景：在一些應用場景中，程序并不需要同步執行，例如用戶注冊之后的郵件或者短信通知提醒。這種場景的實現則是在當前線程，開啟一個新線程，當前線程在開啟新線程之后會繼續往下執行，無需等待新線程執行完成。但例如一些需要延時的場景則不只是開啟新線程執行如此簡單了。譬如提交訂單后在15分鐘內沒有完成支付，訂單需要關閉，這種情況，是否只開啟一個異步線程就不適用了呢。那么就單單實現...

alighters 2019-08-19 10:59 評論0 收藏0
一起來學SpringBoot | 第十三篇：RabbitMQ延遲隊列

摘要：另一種就是用中的位于包下，本質是由和實現的阻塞優先級隊列。表明了一條消息可在隊列中存活的最大時間。當某條消息被設置了或者當某條消息進入了設置了的隊列時，這條消息會在時間后死亡成為。 SpringBoot 是為了簡化 Spring 應用的創建、運行、調試、部署等一系列問題而誕生的產物，自動裝配的特性讓我們可以更好的關注業務本身而不是外部的XML配置，我們只需遵循規范，引入相關的依賴就可...

selfimpr 2019-08-16 12:35 評論0 收藏0
rabbitmq

摘要：一關鍵字和之間的連接關系實際存儲消息。生產者進行接受應答，用來確定這條消息是否正常的發送到了，這種方式也是消息的可靠性投遞的核心保障。支持消息的過期時間，在消息發送時可以進行指定。可以監聽這個隊列中消息做相應的處理。一、rabbitmq關鍵字 Binding：Exchange和Exchange、Queue之間的連接關系Queue：實際存儲消息。Durability：是否持久化，Du...

Hwg 2019-08-19 11:41 評論0 收藏0

發表評論

登陸后可評論

0條評論

honhon

男|高級講師

我要關注我要私信

TA的文章

如何制作數碼管顯示程序

閱讀 2323·2021-10-08 10:04
通配符證書有什麼好處？常用通配符SSL證書有哪些？

閱讀 1097·2021-09-03 10:40
Codepen 每日精選（2018-4-14）

閱讀 1150·2019-08-30 15:53
移動端解決方案學習記錄

閱讀 3309·2019-08-30 13:13
CSS外掛：Sass 之規則(Rules)和指令(Directives)

閱讀 2925·2019-08-30 12:55
拓端數據|避開數據海洋的“暗礁”：識別用戶異常行為

閱讀 2278·2019-08-29 13:21
徹底搞懂js里的__proto__和prototype到底有什么區別？

閱讀 1330·2019-08-26 12:12
前端硬核面試專題之 HTML 24 問

閱讀 2755·2019-08-26 10:37

国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優惠，快來選購！

SpringBoot集成RabbitMQ（死信隊列）

相關文章

SpringBoot+RabbitMq實現延時消息隊列

**一起來學SpringBoot | 第十三篇：RabbitMQ延遲隊列**

rabbitmq

發表評論

0條評論

honhon

男|高級講師

TA的文章

如何制作數碼管顯示程序

通配符證書有什麼好處？常用通配符SSL證書有哪些？

Codepen 每日精選（2018-4-14）

移動端解決方案學習記錄

CSS外掛：Sass 之規則(Rules)和指令(Directives)

拓端數據|避開數據海洋的“暗礁”：識別用戶異常行為

徹底搞懂js里的proto和prototype到底有什么區別？

前端硬核面試專題之 HTML 24 問

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優惠，快來選購！

SpringBoot集成RabbitMQ（死信隊列）

相關文章

發表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優惠，快來選購！