SpringBoot整合RabbitMQ处理死信队列和延迟队列

 更新时间:2022年05月28日 09:23:33   作者:IT利刃出鞘  
这篇文章将通过示例为大家详细介绍SpringBoot整合RabbitMQ时如何处理死信队列和延迟队列,文中的示例代码讲解详细,需要的可以参考一下

简介

说明

本文用示例介绍SpringBoot整合RabbitMQ时如何处理死信队列/延迟队列。

RabbitMQ消息简介

RabbitMQ的消息默认不会超时。 

什么是死信队列?什么是延迟队列?

死信队列:

DLX,全称为Dead-Letter-Exchange,可以称之为死信交换器,也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message)之后,它能被重新被发送到另一个交换器中,这个交换器就是DLX,绑定DLX的队列就称之为死信队列。

以下几种情况会导致消息变成死信:

  • 消息被拒绝(Basic.Reject/Basic.Nack),并且设置requeue参数为false;
  • 消息过期;
  • 队列达到最大长度。

延迟队列:

延迟队列用来存放延迟消息。延迟消息:指当消息被发送以后,不想让消费者立刻拿到消息,而是等待特定时间后,消费者才能拿到这个消息进行消费。

相关网址

详解RabbitMQ中死信队列和延迟队列的使用详解

实例代码

路由配置

package com.example.config;
 
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitRouterConfig {
    public static final String EXCHANGE_TOPIC_WELCOME   = "Exchange@topic.welcome";
    public static final String EXCHANGE_FANOUT_UNROUTE  = "Exchange@fanout.unroute";
    public static final String EXCHANGE_TOPIC_DELAY     = "Exchange@topic.delay";
 
    public static final String ROUTINGKEY_HELLOS        = "hello.#";
    public static final String ROUTINGKEY_DELAY         = "delay.#";
 
    public static final String QUEUE_HELLO              = "Queue@hello";
    public static final String QUEUE_HI                 = "Queue@hi";
    public static final String QUEUE_UNROUTE            = "Queue@unroute";
    public static final String QUEUE_DELAY              = "Queue@delay";
 
    public static final Integer TTL_QUEUE_MESSAGE       = 5000;
 
    @Autowired
    AmqpAdmin amqpAdmin;
 
    @Bean
    Object initBindingTest() {
        amqpAdmin.declareExchange(ExchangeBuilder.fanoutExchange(EXCHANGE_FANOUT_UNROUTE).durable(true).autoDelete().build());
        amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_DELAY).durable(true).autoDelete().build());
        amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_WELCOME)
                .durable(true)
                .autoDelete()
                .withArgument("alternate-exchange", EXCHANGE_FANOUT_UNROUTE)
 
                .build());
 
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HI).build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HELLO)
                .withArgument("x-dead-letter-exchange", EXCHANGE_TOPIC_DELAY)
                .withArgument("x-dead-letter-routing-key", ROUTINGKEY_DELAY)
                .withArgument("x-message-ttl", TTL_QUEUE_MESSAGE)
                .build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_UNROUTE).build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_DELAY).build());
 
        amqpAdmin.declareBinding(new Binding(QUEUE_HELLO, Binding.DestinationType.QUEUE,
                EXCHANGE_TOPIC_WELCOME, ROUTINGKEY_HELLOS, null));
        amqpAdmin.declareBinding(new Binding(QUEUE_UNROUTE, Binding.DestinationType.QUEUE,
                EXCHANGE_FANOUT_UNROUTE, "", null));
        amqpAdmin.declareBinding(new Binding(QUEUE_DELAY, Binding.DestinationType.QUEUE,
                EXCHANGE_TOPIC_DELAY, ROUTINGKEY_DELAY, null));
 
        return new Object();
    }
}

控制器

package com.example.controller;
 
import com.example.config.RabbitRouterConfig;
import com.example.mq.Sender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.time.LocalDateTime;
 
@RestController
public class HelloController {
    @Autowired
    private Sender sender;
 
    @PostMapping("/hi")
    public void hi() {
        sender.send(RabbitRouterConfig.QUEUE_HI, "hi1 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/hello1")
    public void hello1() {
        sender.send("hello.a", "hello1 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/hello2")
    public void hello2() {
        sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "hello.b", "hello2 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/ae")
    public void aeTest() {
        sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "nonono", "ae message:" + LocalDateTime.now());
    }
}

发送器

package com.example.mq;
 
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import java.util.Date;
 
@Component
public class Sender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
 
    public void send(String routingKey, String message) {
        this.rabbitTemplate.convertAndSend(routingKey, message);
    }
 
    public void send(String exchange, String routingKey, String message) {
        this.rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }
}

接收器

package com.example.mq;
 
import com.example.config.RabbitRouterConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
public class Receiver {
    @RabbitListener(queues = RabbitRouterConfig.QUEUE_HI)
    public void hi(String payload) {
        System.out.println ("Receiver(hi) : "  + payload);
    }
 
    // @RabbitListener(queues = RabbitRouterConfig.QUEUE_HELLO)
    // public void hello(String hello) throws InterruptedException {
    //     System.out.println ("Receiver(hello) : "  + hello);
    //     Thread.sleep(5 * 1000);
    //     System.out.println("(hello):sleep over");
    // }
    //
    // @RabbitListener(queues = RabbitRouterConfig.QUEUE_UNROUTE)
    // public void unroute(String hello) throws InterruptedException {
    //     System.out.println ("Receiver(unroute) : "  + hello);
    //     Thread.sleep(5 * 1000);
    //     System.out.println("(unroute):sleep over");
    // }
 
    @RabbitListener(queues = RabbitRouterConfig.QUEUE_DELAY)
    public void delay(String hello) throws InterruptedException {
        System.out.println ("Receiver(delay) : "  + hello);
        Thread.sleep(5 * 1000);
        System.out.println("(delay):sleep over");
    }
}

application.yml

server:
#  port: 9100
  port: 9101
spring:
  application:
#    name: demo-rabbitmq-sender
    name: demo-rabbitmq-receiver
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: 123456
#    virtualHost: /
    publisher-confirms: true
    publisher-returns: true
#    listener:
#      simple:
#        acknowledge-mode: manual
#      direct:
#        acknowledge-mode: manual

实例测试

分别启动发送者和接收者。

访问:http://localhost:9100/hello2

五秒钟后输出:

Receiver(delay) : hello2 message:2020-11-27T09:30:51.548
(delay):sleep over

以上就是SpringBoot整合RabbitMQ处理死信队列和延迟队列的详细内容,更多关于SpringBoot RabbitMQ死信队列 延迟队列的资料请关注脚本之家其它相关文章!

相关文章

  • 如何在SpringBoot 中使用 Druid 数据库连接池

    如何在SpringBoot 中使用 Druid 数据库连接池

    这篇文章主要介绍了SpringBoot 中使用 Druid 数据库连接池的实现步骤,帮助大家更好的理解和学习使用SpringBoot,感兴趣的朋友可以了解下
    2021-03-03
  • Spring Cloud实现提供API给客户端的方法详解

    Spring Cloud实现提供API给客户端的方法详解

    这篇文章主要给大家介绍了关于Spring Cloud实现提供API给客户端的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧。
    2018-01-01
  • java 导入Excel思路及代码示例

    java 导入Excel思路及代码示例

    这篇文章主要介绍了java 导入Excel思路及代码示例,简要分析了设计思路,然后分享了实例代码,具有一定参考价值,需要的朋友可以了解下。
    2017-11-11
  • 解决Elasticsearch因jdk版本问题启动失败的问题

    解决Elasticsearch因jdk版本问题启动失败的问题

    这篇文章主要介绍了解决Elasticsearch因jdk版本问题启动失败的问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-07-07
  • SpringBoot接口如何对参数进行校验

    SpringBoot接口如何对参数进行校验

    这篇文章主要介绍了SpringBoot接口如何对参数进行校验,在以SpringBoot开发Restful接口时, 对于接口的查询参数后台也是要进行校验的,同时还需要给出校验的返回信息放到上文我们统一封装的结构中
    2022-07-07
  • 详解java的值传递、地址传递、引用传递

    详解java的值传递、地址传递、引用传递

    这篇文章主要介绍了详解java的值传递、地址传递、引用传递的相关资料,需要的朋友可以参考下
    2017-06-06
  • Java 图片压缩实现思路及代码

    Java 图片压缩实现思路及代码

    本文为大家详细介绍下图片压缩的具体实现思路及java代码,想学习的各位可以参考下哈,希望对大家有所帮助
    2013-07-07
  • 详解Spring batch 入门学习教程(附源码)

    详解Spring batch 入门学习教程(附源码)

    本篇文章主要介绍了Spring batch 入门学习教程(附源码),小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-11-11
  • response对象的使用(实例讲解)

    response对象的使用(实例讲解)

    下面小编就为大家带来一篇response对象的使用(实例讲解)。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-09-09
  • mybatis-plus主键id生成、字段自动填充的实现代码

    mybatis-plus主键id生成、字段自动填充的实现代码

    这篇文章主要介绍了mybatis-plus主键id生成、字段自动填充的实现代码,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-12-12

最新评论