网站的管理,在线包装设计软件,网络流量统计工具,文章修改网站一、简介
在rabbitMQ中常用的交换机有三种#xff0c;直连交换机、广播交换机、主题交换机#xff1b;
直连交换机中队列与交换机需要约定好routingKey去进行绑定#xff1b;
广播交换机并不需要routingKey绑定,只需队列与交换机绑定即可#xff1b;
主题交换机最大的特…一、简介
在rabbitMQ中常用的交换机有三种直连交换机、广播交换机、主题交换机
直连交换机中队列与交换机需要约定好routingKey去进行绑定
广播交换机并不需要routingKey绑定,只需队列与交换机绑定即可
主题交换机最大的特点可以通过*和#去匹配队列 而死信队列其实就是平常的队列的一种通常我会使用直连交换机来作为死信队列所以说死信队列其实就是我们在处理业务中慢慢衍生出来的一个名词、一种方案它和普通的队列是一样的。
二、使用场景 我们知道在使用队列时有几种应答模式,比如自动应答auto、手动应答manual等而在使用自动应答时无论消息是否成功消费达到重试次数后就会自动的把此消息给删除掉了当然我们是不想把没有消费成功的消息给删除掉的。而开启手动应答时配置的重试机制会失效 当有消费失败的消息时 会进入死循环。 那么为了解决此场景就引入了死信队列。当有不能正常消费的消息时 就把此消息给打到死信队列中然后再根据实际情况去处理此信息。
关于自动应答和手动应答可参考这篇博客
rabbitMQ手动应答与自动应答_骑着蜗牛打天下的博客-CSDN博客 在 RabbitMQ 中充当主角的就是消息在不同场景下消息会有不同地表现。
死信就是消息在特定场景下的一种表现形式这些场景包括
1. 消息被拒绝访问即 RabbitMQ返回 basicNack 的信号时。 或者拒绝basicReject
2. 消费者发生异常超过重试次数 。
3. 消息的 TTL 过期时。
4. 消息队列达到最大长度。 三、代码实现 父pom文件
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.1/version
!-- version2.2.5.RELEASE/version--relativePath/ !-- lookup parent from repository --/parentgroupIdcom.chensir/groupIdartifactIdspring-boot-rabbitmq/artifactIdversion0.0.1-SNAPSHOT/versionnamespring-boot-rabbitmq/namepropertiesjava.version8/java.versionhutool.version5.8.3/hutool.versionlombok.version1.18.24/lombok.version/propertiesdescriptionspring-boot-rabbitmq/descriptionpackagingpom/packagingmodulesmoduledirect-exchange/modulemodulefanout-exchange/modulemoduletopic-exchange/modulemodulegame-exchange/modulemoduledead-letter-queue/modulemoduledelay-queue/modulemoduledelay-queue2/module/modulesdependencyManagementdependenciesdependencygroupIdcn.hutool/groupIdartifactIdhutool-all/artifactIdversion${hutool.version}/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion${lombok.version}/version/dependency/dependencies/dependencyManagement/projectpom文件
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdcom.chensir/groupIdartifactIdspring-boot-rabbitmq/artifactIdversion0.0.1-SNAPSHOT/versionrelativePath../pom.xml/relativePath/parentartifactIddead-letter-queue/artifactIddependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdcn.hutool/groupIdartifactIdhutool-all/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-rabbit-test/artifactIdscopetest/scope/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project配置文件
server.port8084
#host
spring.rabbitmq.host121.40.100.66
#默认5672
spring.rabbitmq.port5672
#用户名
spring.rabbitmq.usernameguest
#密码
spring.rabbitmq.passwordguest
#连接到代理时用的虚拟主机
spring.rabbitmq.virtual-host/
#每个消费者每次可最大处理的nack消息数量
spring.rabbitmq.listener.simple.prefetch1
#表示消息确认方式其有三种配置方式分别是none、manual(手动)和auto(自动)默认auto
spring.rabbitmq.listener.simple.acknowledge-modeauto
#监听重试是否可用
spring.rabbitmq.listener.simple.retry.enabledtrue
#最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts5
#最大重试时间间隔
spring.rabbitmq.listener.simple.retry.max-interval3000ms
#第一次和第二次尝试传递消息的时间间隔
spring.rabbitmq.listener.simple.retry.initial-interval1000ms
#应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.multiplier2
#决定被拒绝的消息是否重新入队默认是true与参数acknowledge-mode有关系
spring.rabbitmq.listener.simple.default-requeue-rejectedfalseconfig
正常队列config
package com.chensir.config;import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class RabbitConfig {Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}Beanpublic DirectExchange directExchange(){return new DirectExchange(DirectExchange,true,false);}Beanpublic Queue directQueueLong(){return QueueBuilder.durable(DirectQueue).deadLetterExchange(DeadLetterExchange).deadLetterRoutingKey(dead)//20s还没消费就打到死信队列中.ttl(20000)//当队列中长度有500个消息,也打入死信队列.maxLength(500).build();}Beanpublic Binding binding(){return BindingBuilder.bind(directQueueLong()).to(directExchange()).with(direct123);}
}
死信队列config
package com.chensir.config;import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 死信队列 一般由运维在rebbitMQ服务创建交换机和队列 不需要代码配置*/
//Configuration
public class DeadLetterConfig {Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}Beanpublic DirectExchange directExchange() {DirectExchange directExchange new DirectExchange(DeadLetterExchange);return directExchange;}Beanpublic Queue queue() {Queue deadLetterQueue QueueBuilder.durable(DeadLetterQueue).build();return deadLetterQueue;}Beanpublic Binding binding() {Binding binding BindingBuilder.bind(queue()).to(directExchange()).with(dead);return binding;}}生产者
package com.chensir.provider;import cn.hutool.json.JSONUtil;
import com.chensir.model.OrderIngOk;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;Component
public class DirectProvider {Resourceprivate RabbitTemplate rabbitTemplate;public void send(){// 死信队列
// rabbitTemplate.convertAndSend(DeadLetterExchange, dead,123);for (int i1;i7;i){OrderIngOk orderIngOk new OrderIngOk();orderIngOk.setOrderNo(202308289687-i);orderIngOk.setId(i);orderIngOk.setUserName(倪海杉);
// String s JSONUtil.toJsonStr(orderIngOk);rabbitTemplate.convertAndSend(DirectExchange, direct123,orderIngOk);}}}消费者
正常队列消费者
package com.chensir.consumer;import cn.hutool.json.JSONUtil;
import com.chensir.model.OrderIngOk;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;Component
public class DirectConsumer {RabbitHandlerRabbitListener(queues DirectQueue )public void process(OrderIngOk orderIngOk) throws IOException {try {// 处理业务开始if(orderIngOk.getId().equals(5)){int a 0;int b 2/a;}System.out.println(接受到消息,并正常处理结束 JSONUtil.toJsonStr(orderIngOk));} catch (Exception ex){System.out.println(ex.getMessage());System.out.println(接受到消息,发生异常 JSONUtil.toJsonStr(orderIngOk));//自动应答 当消费者成功消费消息时会自动把消息删除,而没有成功消费消息时需要给重试机制抛出个异常 重试机制才会开启重试throw ex;//手动模式//channel.basicReject(deliveryTag,true);//channel.basicNack(deliveryTag,false,true);}}
}
死信队列消费者
package com.chensir.consumer;import com.chensir.model.OrderIngOk;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class DeadConsumer {RabbitHandlerRabbitListener(queues DeadLetterQueue)public void process(OrderIngOk orderIngOk) {System.out.println(这条信息在运行时发生了未知的异常此信息被打到了死信队列,被死信队列消费者消费成功orderIngOk);}
}结果
接受到消息,并正常处理结束{id:1,OrderNo:202308289687-1,userName:倪海杉}
接受到消息,并正常处理结束{id:2,OrderNo:202308289687-2,userName:倪海杉}
接受到消息,并正常处理结束{id:3,OrderNo:202308289687-3,userName:倪海杉}
接受到消息,并正常处理结束{id:4,OrderNo:202308289687-4,userName:倪海杉}
/ by zero
接受到消息,发生异常{id:5,OrderNo:202308289687-5,userName:倪海杉}
/ by zero
接受到消息,发生异常{id:5,OrderNo:202308289687-5,userName:倪海杉}
/ by zero
接受到消息,发生异常{id:5,OrderNo:202308289687-5,userName:倪海杉}
/ by zero
接受到消息,发生异常{id:5,OrderNo:202308289687-5,userName:倪海杉}
/ by zero
接受到消息,发生异常{id:5,OrderNo:202308289687-5,userName:倪海杉}
2023-08-28 16:45:39.848 WARN 24432 --- [ntContainer#1-1] o.s.a.r.r.RejectAndDontRequeueRecoverer : Retries exhausted for message (Body:[B1a6e663a(byte[58]) MessageProperties [headers{__TypeId__com.chensir.model.OrderIngOk}, contentTypeapplication/json, contentEncodingUTF-8, contentLength0, receivedDeliveryModePERSISTENT, priority0, redeliveredfalse, receivedExchangeDirectExchange, receivedRoutingKeydirect123, deliveryTag5, consumerTagamq.ctag-f9Up1UES-F3rDvb-AK16xw, consumerQueueDirectQueue])这条信息在运行时发生了未知的异常此信息被打到了死信队列,被死信队列消费者消费成功OrderIngOk(id5, OrderNo202308289687-5, userName倪海杉)
接受到消息,并正常处理结束{id:6,OrderNo:202308289687-6,userName:倪海杉}
