志愿海南网站,新乡手机网站建设哪家好,福州专业制作网站,温州平阳县网站建设兼职springboot整合rabbitmq死信队列
什么是死信
说道死信#xff0c;可能大部分观众大姥爷会有懵逼的想法#xff0c;什么是死信#xff1f;死信队列#xff0c;俗称DLX#xff0c;翻译过来的名称为Dead Letter Exchange 死信交换机。当消息限定时间内未被消费#xff0c;…springboot整合rabbitmq死信队列
什么是死信
说道死信可能大部分观众大姥爷会有懵逼的想法什么是死信死信队列俗称DLX翻译过来的名称为Dead Letter Exchange 死信交换机。当消息限定时间内未被消费成为 Dead Message后可以被重新发送到另一个交换机中发挥其应有的价值需要测试死信队列则需要先梳理整体的思路如可以采取如下方式进行配置
从上面的逻辑图中可以发现大致的思路
.1. 消息队列分为正常交换机、正常消息队列以及死信交换机和死信队列。 2. 正常队列针对死信信息需要将数据 重新 发送至死信交换机中。
死信使用的场景
消息被拒绝消息ttl过期队列达到最大长度
这三种场景就会成为死信然后放入死信交换机
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;Configuration
public class RabbitmqConfig {//正常交换机的名字public final static String EXCHANGE\_NAME exchange\_name;//正常队列的名字public final static String QUEUE\_NAMEqueue\_name;//死信交换机的名字public final static String EXCHANGE\_DEAD exchange\_dead;//死信队列的名字public final static String QUEUE\_DEADqueue\_dead;//死信路由keypublic final static String DEAD\_KEYdead.key;//创建正常交换机Bean(EXCHANGE\_NAME)public Exchange exchange(){return ExchangeBuilder.topicExchange(EXCHANGE\_NAME)//持久化 mq重启后数据还在.durable(true).build();}//创建正常队列Bean(QUEUE\_NAME)public Queue queue(){//正常队列和死信进行绑定 转发到 死信队列配置参数MapString,ObjectmapgetMap();return new Queue(QUEUE\_NAME,true,false,false,map);}//正常队列绑定正常交换机 设置规则 执行绑定 定义路由规则 requestmaping映射Beanpublic Binding binding(Qualifier(QUEUE\_NAME) Queue queue,Qualifier(EXCHANGE\_NAME) Exchange exchange){return BindingBuilder.bind(queue).to(exchange)//路由规则.with(app.#).noargs();}//创建死信队列Bean(QUEUE\_DEAD)public Queue queueDead(){return new Queue(QUEUE\_DEAD,true,false,false);}//创建死信交换机Bean(EXCHANGE\_DEAD)public Exchange exchangeDead(){return ExchangeBuilder.topicExchange(EXCHANGE\_DEAD).durable(true) //持久化 mq重启后数据还在.build();}//绑定死信队列和死信交换机Beanpublic Binding deadBinding(){return BindingBuilder.bind(queueDead()).to(exchangeDead())//路由规则 正常路由key.with(DEAD\_KEY).noargs();}/\*\*获取死信的配置信息\*\*\*/public MapString,ObjectgetMap(){//3种方式 任选其一,选择其他方式之前,先把交换机和队列删除了,在启动项目,否则报错。//方式一MapString,Object mapnew HashMap(16);//死信交换器名称过期或被删除因队列长度超长或因空间超出阈值的消息可指定发送到该交换器中map.put(x-dead-letter-exchange, EXCHANGE\_DEAD);//死信消息路由键在消息发送到死信交换器时会使用该路由键如果不设置则使用消息的原来的路由键值map.put(x-dead-letter-routing-key, DEAD\_KEY);//方式二//消息的过期时间单位毫秒达到时间 放入死信队列// map.put(x-message-ttl,5000);//方式三//队列最大长度超过该最大值则将从队列头部开始删除消息放入死信队列一条数据// map.put(x-max-length,3);return map;}}配置文件信息
spring:rabbitmq:host: 192.168.23.135port: 5672username: adminpassword: admin#虚拟主机virtual-host: dmg-alistener:simple:#自动ackacknowledge-mode: autoretry:#最大重试次数max-attempts: 3#开启重试enabled: true引入 rabbitmq 依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency生产者
RestController
RequestMapping(p)
public class TestController {Autowiredprivate RabbitTemplate rabbitTemplate;GetMapping(/test)public String test(){//正常交换机 正常路由键 正常消息内容rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE\_NAME,app.test,我是生产者);return aa;}
}//消费者
Component
public class Xf {//监听正常队列名称RabbitListener(queues {RabbitmqConfig.QUEUE\_NAME})public void normal(String payload, Message message, Channel channel) throws IOException {System.out.println(正常消息:payload);long tagmessage.getMessageProperties().getDeliveryTag();try{// int i1/0;//手动签收channel.basicAck(tag,true);}catch (RuntimeException runtimeException){//出现异常 删除消息 放入死信队列channel.basicReject(tag,false);}}监听死信队列名称 RabbitListener(queues {RabbitmqConfig.QUEUE\_DEAD})public void dead(String payload, Message message, Channel channel) throws IOException {System.out.println(死信队列:payload);//删除消息 放入数据库 人工处理long deliveryTagmessage.getMessageProperties().getDeliveryTag();channel.basicAck(deliveryTag,true);}
}
