网站开发的论文引言,微信网站开发视频教程,企业网页建设公司联系电话,高校门户网站建设方案死信队列
RabbitMQ 的死信队列#xff08;Dead-Letter-Exchanges#xff0c;简称 DLX#xff09;是一个强大的特性#xff0c;它允许在消息在队列中无法被正常消费#xff08;例如#xff0c;消息被拒绝并且没有设置重新入队#xff0c;或者消息过期#xff09;时Dead-Letter-Exchanges简称 DLX是一个强大的特性它允许在消息在队列中无法被正常消费例如消息被拒绝并且没有设置重新入队或者消息过期时将这些消息转发到另一个交换机。这个特性在很多场景下都非常有用比如重试机制、延迟队列等。
以下是关于 RabbitMQ 死信队列的一些要点
如何设置死信队列
要设置死信队列你需要在队列声明时指定几个参数
x-dead-letter-exchange指定消息在变为死信后要发送到的交换机。 x-dead-letter-routing-key可选指定消息在变为死信后使用的路由键。如果未设置则使用原消息的路由键。 message-ttl 或 x-message-ttl可选设置消息的生存时间TTL。当消息在队列中的时间超过此值后它将成为死信。 x-max-length可选设置队列的最大长度。当队列中的消息数量超过此值时最早的消息将成为死信。 这些参数可以通过 RabbitMQ 的管理界面、命令行工具或编程 API 设置。
死信队列的工作流程
当一个消息在队列中由于某些原因如过期、被拒绝且未设置重新入队、队列达到最大长度等成为死信时。 RabbitMQ 会检查该队列是否配置了 x-dead-letter-exchange。 如果配置了RabbitMQ 会将死信发送到指定的死信交换机。 死信交换机再根据配置的路由键或原消息的路由键将消息路由到相应的队列。
windows安装RabbitMQ
RabbitMQ是建立在强大的Erlang OTP平台上因此安装Rabbit MQ的前提是安装Erlang。
安装Erlang
因为RabbitMQ服务器是用Erlang语言编写的 所以你需要去查看rabbitMq适应Erlang的版本因为不同的rabbitMq版本对应不同的Erlang版本可以点击如下该链接查看版本匹配度 https://www.rabbitmq.com/which-erlang.html#compatibility-matrix
下载地址Erlang 推荐使用链接: https://download.csdn.net/download/weixin_42123075/89064540这里包含Erlang和对应版本的RabbitMQ。 下载完成后先安装Erlang。
RabbitMQ安装
下载地址 https://github.com/rabbitmq/rabbitmq-server/releases?page7
设置环境变量
windows R 在左下角的弹出框中输入control system点击高级系统设置点击环境变量系统变量新建变量名RABBITQM_SERVER 变量值rabbitmq安装地址bin目录之前的地址比如我的是D:\Software\rabbitmq\rabbitmq_server-3.8.15找到系统变量中的PATH点击编辑点击新建输入%RABBITQM_SERVER%\sbin点击确定。
安装 rabbitmq_management
打开Windows Poweshell或者cmd输入如下指令
rabbitmq-plugins list如下图所示
安装rabbitmq_management插件 这款插件是可以可视化的方式查看RabbitMQ服务器实例的状态以及操控RabbitMQ服务器。
rabbitmq-plugins enable rabbitmq_management
如下图所示
启动rabbitMq服务
安装rabbitMq的目录(我的是D:\Software\rabbitmq\rabbitmq_server-3.8.15) - sbin目录 - 双击rabbitmq-server.bat我如下图所示
如果启动报错 原因是原先的RabbitMq数据与新的RabbitMq有些不兼容 解决方案 sudo rm -rf /var/lib/rabbitmq/mnesia win解决方案在 RabbitMQ 安装目录中找到 mnesia 目录通常在 C:\Users\AppData\Roaming\RabbitMQ\db然后删除其中的文件 参考 https://www.cnblogs.com/gaoyuechen/p/10368188.html,重新启动。 启动成功但程序不是以deamon形式运行。只要执行-detached可执行成功。
rabbitmq-server -detached访问http://localhost:15672/管理页面用户名密码默认均为guest。
创建死信队列和交换机
创建dead-letter-queue死信队列 创建dead-letter-exchange死信交换机并绑定死信队列 创建普通队列和普通交换机并指定死信交换机
创建普通队列myQueue和普通交换机myExchange交换机类型为Topic并指定死信队列
SpringBoot集成RabbitMQ
添加maven依赖
!-- RabbitMQ --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency在yml配置文件中的spring下添加RabbitMQ的连接信息
spring:rabbitmq:addresses: localhost:5672connection-timeout: 15000password: guestusername: guest# 使用启用消息确认模式
# publisher-confirms: truevirtual-host: /RabbitMQConfig import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class RabbitMQConfig {// 普通队列public final static String QUEUE_NAME myQueue;// 普通交换机public final static String EXCHANGE_NAME myExchange;// 普通队列路由public final static String ROUTING_KEY myRoutingKey;// 死信交换机public static final String DEAD_LETTER_EXCHANGE dead-letter-exchange;// 死信队列public static final String DEAD_LETTER_QUEUE dead-letter-queue;// 死信路由public static final String DEAD_LETTER_ROUTING_KEY dead-letter-key;BeanTopicExchange myExchange() {return new TopicExchange(EXCHANGE_NAME);}BeanBinding binding(Queue myQueue, TopicExchange myExchange) {return BindingBuilder.bind(myQueue).to(myExchange).with(ROUTING_KEY);}/*** 定义死信交换机* return DirectExchange*/Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange(DEAD_LETTER_EXCHANGE);}/*** 定义死信队列* return Queue*/Beanpublic Queue deadLetterQueue() {return new Queue(DEAD_LETTER_QUEUE,true,false,false,null);}/*** 死信队列绑定死信交换机* param deadLetterQueue 死信队列* param deadLetterExchange 死信交换机* return Binding*/Beanpublic Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with(DEAD_LETTER_ROUTING_KEY);}/*** 普通队列声明指定死信交换机* return*/Beanpublic Queue myQueue() {return QueueBuilder.durable(QUEUE_NAME)// 设置死信交换机.withArgument(x-dead-letter-exchange, DEAD_LETTER_EXCHANGE)// 设置死信路由键.withArgument(x-dead-letter-routing-key, DEAD_LETTER_ROUTING_KEY)// 设置队列最大长度// .withArgument(x-max-length, 5).build();}}发送消息类RabbitMQService
import com.ruoyi.quartz.config.RabbitMQConfig;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;Service
public class RabbitMQService {Autowiredprivate AmqpTemplate rabbitTemplate;public void send(String message) {rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, message);}/*** 发送消息并设置过期时间* param exchange 交换机* param routingKey 路由* param messageBody 消息体* param expirationTimeInMillis 过期时间单位毫秒*/public void sendMessageWithExpiration(String exchange, String routingKey, String messageBody, int expirationTimeInMillis) {MessageProperties properties new MessageProperties();// 设置消息的过期时间properties.setExpiration(String.valueOf(expirationTimeInMillis));Message message MessageBuilder.withBody(messageBody.getBytes()).andProperties(properties).build();rabbitTemplate.convertAndSend(exchange, routingKey, message);}
}监听队列接收消息ReceiverService import com.ruoyi.quartz.config.RabbitMQConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;import java.text.SimpleDateFormat;
import java.util.Date;Service
public class ReceiverService {/*** 监听队列消息 如果要测试死信队列就不要监听此队列* param message 消息*/
// RabbitListener(queues RabbitMQConfig.QUEUE_NAME)public void receive(String message) {System.out.println(Received message );}/*** 监听死信队列* param message*/RabbitListener(queues RabbitMQConfig.DEAD_LETTER_QUEUE)public void processDeadLetter(String message) {String time new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date());System.out.println(Received dead letter message: message 当前时间 time);// 处理死信队列中的消息}
}测试死信队列 GetMapping(/sendMessageTtl/{message})public void sendMessageTtl(PathVariable String message){log.info(当前时间发送:{},发送5条消息给两个TTL队列:{},new Date().toString(),message);for (int i 0; i 6; i) {System.out.println(测试延迟队列DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS,new Date()));rabbitMQService.sendMessageWithExpiration(RabbitMQConfig.EXCHANGE_NAME,RabbitMQConfig.ROUTING_KEY,测试5秒延迟》,5000);}}5秒后私信队列收到消息
