rabbitmq之死信队列（四）

概念死信队列顾名思义就是无法被消费的消息某些时候由于特定的原因导致 queue 中的某些消息无法被消费比如被消费者Nack掉了并且消费者绑定了死信队列那么被丢弃的就会到私信队列中。死信的来源前提普通队列绑定了死信队列那么有以下来源前提消费者绑定了死信队列消息 TTL 过期队列达到最大长度(队列满了无法再添加数据到 mq 中)消息被拒绝(basic.reject 或 basic.nack)并且 requeuefalse代码演示ttl过期效果模拟下图关系添加依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency设置配置文件# 服务端口 spring.rabbitmq.host192.168.136.128spring.rabbitmq.port5672spring.rabbitmq.usernameadmin spring.rabbitmq.passwordadmindefault-requeue-rejected要改为false代表消费方拒绝消息后不会重新放回普通队列里。创建生产者ComponentpublicclassProducer{AutowiredprivateRabbitTemplaterabbitTemplate;// 2: 路由keyprivateStringrouteKeyzhangsan;publicvoidmakeOrder(){// 发送消息给所有客户for(inti1;i10;i){Stringcontent消息i;System.out.println(生产者发送消息:i)r abbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE,routeKey,content,message-{//设置消息过期时间10s,如果十秒内没有被消费者接受就自动转入死信队列message.getMessageProperties().setExpiration(10000);retrun message;});}}}创建配置类ConfigurationpublicclassRabbitConfig{//普通交换机名称publicstaticfinalStringNORMAL_EXCHANGEnormal_exchange;//死信交换机名称publicstaticfinalStringDEAD_EXCHANGEdead_exchange;//普通队列名称publicstaticfinalStringNORMAL_QUEUEnormal_queue;//死信死信队列名称publicstaticfinalStringDEAD_QUEUEdead_queue;//声明队列Bean(normalQueueDeclared)publicQueuenormalQueueDeclared(){//正常队列绑定死信队列信息MapString,ObjectparamsnewHashMap();//正常队列设置死信交换机 参数 key 是固定值params.put(x-dead-letter-exchange,DEAD_EXCHANGE);//正常队列设置死信 routing-key 参数 key 是固定值params.put(x-dead-letter-routing-key,lisi);returnQueueBuilder.durable(NORMAL_QUEUE).withArguments(params).build();}//声明死信队列Bean(deadlQueueDeclared)publicQueuedeadlQueueDeclared(){returnQueueBuilder.durable(DEAD_QUEUE).build();}//声明普通交换机Bean(myNormalExchange)publicDirectExchangemyNormalExchange(){returnnewDirectExchange(NORMAL_EXCHANGE);}//声明死信交换机Bean(myDeadExchange)publicDirectExchangemyDeadExchange(){returnnewDirectExchange(DEAD_EXCHANGE);}//绑定将队列和交换机绑定BeanpublicBindingbinding1(Qualifier(normalQueueDeclared)QueuenormalQueueDeclared,Qualifier(myNormalExchange)DirectExchangemyNormalExchange){returnBindingBuilder.bind(normalQueueDeclared).to(myNormalExchange).with(zhangsan);}BeanpublicBindingbinding2(Qualifier(deadlQueueDeclared)QueuedeadlQueueDeclared,Qualifier(myDeadExchange)DirectExchangemyDeadExchange){returnBindingBuilder.bind(deadlQueueDeclared).to(myDeadExchange).with(lisi);}}消费方单独建立一个boot工程只启动生产者不启动消费方模拟消息过期的效果 c1// bindings用来确定队列和交换机绑定关系RabbitListener(bindingsQueueBinding(//指定获取信息的队列valueQueue(valueRabbitConfig.NORMAL_QUEUE,autoDeletefalse),// 指定交换机exchangeExchange(valueRabbitConfig.NORMAL_EXCHANGE,// 这里是确定的交换机的模式typeExchangeTypes.DIRECT)))ComponentpublicclassConsumer01{// RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值RabbitHandlerpublicvoidmessagerevice(Stringmessage){// 此处省略发邮件的逻辑System.out.println(c1--------------message);}}消费方c2// bindings用来确定队列和交换机绑定关系RabbitListener(bindingsQueueBinding(//指定获取信息的队列valueQueue(valueRabbitConfig.DEAD_QUEUE,autoDeletefalse),// 指定交换机exchangeExchange(valueRabbitConfig.DEAD_EXCHANGE,// 这里是确定的交换机的模式typeExchangeTypes.DIRECT)))ComponentpublicclassConsumer02{// RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值RabbitHandlerpublicvoidmessagerevice(Stringmessage){// 此处省略发邮件的逻辑System.out.println(c2--------------message);}}观察结果生产者发送了10条消息因为消费方没有启动所以显示10条消息未消费10秒后消息过期自动加入到死信队列启动消费方c2,此时就可以看到死信队列中的数据已经被c2消费掉了c2--------------消息1c2--------------消息2c2--------------消息3c2--------------消息4c2--------------消息5c2--------------消息6c2--------------消息7c2--------------消息8c2--------------消息9c2--------------消息10队列达到最大长度代码演示生产者取消发送消息的ttlComponentpublicclassProducer{AutowiredprivateRabbitTemplaterabbitTemplate;// 2: 路由keyprivateStringrouteKeyzhangsan;publicvoidmakeOrder(){// 发送消息给所有客户for(inti1;i10;i){Stringcontent消息i;System.out.println(生产者发送消息:i)r abbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE,routeKey,content,null);}}}设置普通队列的最大长度//声明队列Bean(normalQueueDeclared)publicQueuenormalQueueDeclared(){//正常队列绑定死信队列信息MapString,ObjectparamsnewHashMap();//正常队列设置死信交换机 参数 key 是固定值params.put(x-dead-letter-exchange,DEAD_EXCHANGE);//正常队列设置死信 routing-key 参数 key 是固定值params.put(x-dead-letter-routing-key,lisi);//设置队列的最大长度params.put(x-max-length,6);NORMAL_QUEUEreturnQueueBuilder.durable(NORMAL_QUEUE).withArguments(params).build();}//设置队列的最大长度一旦超过最大长度以外的消息都会路由到死信队列params.put(“x-max-length”, 6);观察结果不启动消费者否则队列中的消息会被及时消费很难演示出效果。普通队列我们设置了最大长度6所以可以看到有4条消息路由到了死信队列。消费者拒绝消息代码演示c1设置拒绝消费// bindings用来确定队列和交换机绑定关系RabbitListener(bindingsQueueBinding(//指定获取信息的队列valueQueue(valueRabbitConfig.NORMAL_QUEUE,autoDeletefalse),// 指定交换机exchangeExchange(valueRabbitConfig.NORMAL_EXCHANGE,// 这里是确定的交换机的模式typeExchangeTypes.DIRECT)))ComponentpublicclassConsumer01{// RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值RabbitHandlerpublicvoidmessagerevice(Channelchannel,Messagemessage){StringcontentnewString(message.getBody())System.out.println(消费者 content);if(消息1.equals(content)){//requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中System.out.println(C1 接收到消息content并拒绝签收该消息);channel.basicReject(message.getMessageProperties().getDeliveryTag(),false)}// 此处省略发邮件的逻辑System.out.println(c1--------------message);}}启动生产者发消息启动c1消费消息因为c1拒绝了一个消息所以该消息被路由到了死信队列中。死信队列的应用场景一般用在较为重要的业务队列中确保未被正确消费的消息不被丢弃一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常处理过程中参数校验异常或者因网络波动导致的查询异常等等通过配置死信队列可以让未正确处理的消息暂存到另一个队列中待后续排查清楚问题后编写相应的处理代码来处理死信消息这样比手工恢复数据要好太多了。总结死信队列其实并没有什么特别不过是绑定在死信交换机上的普通队列而死信交换机也只是一个普通的交换机不过是用来专门处理死信的交换机。死信队列之延时队列延时队列链接