今天小编给大家分享一下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");
}
}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处理死信队列和延迟队列”这篇文章的所有内容，感谢各位的阅读！相信大家阅读完这篇文章都有很大的收获，小编每天都会为大家更新不同的知识，如果还想学习更多的知识，请关注亿速云行业资讯频道。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
