企业官网建设流程全解析

概念

死信队列，顾名思义就是无法被消费的消息，某些时候由于特定的原因导致 queue 中的某些消息无法被消费，比如被消费者Nack掉了并且消费者绑定了死信队列，那么被丢弃的就会到私信队列中。

死信的来源

前提普通队列绑定了死信队列那么有以下来源，前提消费者绑定了死信队列

• 消息 TTL 过期
• 队列达到最大长度(队列满了，无法再添加数据到 mq 中)
• 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

代码演示ttl过期效果

模拟下图关系
添加依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

设置配置文件

# 服务端口 spring.rabbitmq.host=192.168.136.128spring.rabbitmq.port=5672spring.rabbitmq.username=admin spring.rabbitmq.password=admin

default-requeue-rejected要改为false，代表消费方拒绝消息后不会重新放回普通队列里。

创建生产者

@ComponentpublicclassProducer{@AutowiredprivateRabbitTemplaterabbitTemplate;// 2: 路由keyprivateStringrouteKey="zhangsan";publicvoidmakeOrder(){// 发送消息给所有客户for(inti=1;i<=10;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_EXCHANGE="normal_exchange";//死信交换机名称publicstaticfinalStringDEAD_EXCHANGE="dead_exchange";//普通队列名称publicstaticfinalStringNORMAL_QUEUE="normal_queue";//死信死信队列名称publicstaticfinalStringDEAD_QUEUE="dead_queue";//声明队列@Bean("normalQueueDeclared")publicQueuenormalQueueDeclared(){//正常队列绑定死信队列信息Map<String,Object>params=newHashMap<>();//正常队列设置死信交换机 参数 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(bindings=@QueueBinding(//指定获取信息的队列value=@Queue(value=RabbitConfig.NORMAL_QUEUE,autoDelete="false"),// 指定交换机exchange=@Exchange(value=RabbitConfig.NORMAL_EXCHANGE,// 这里是确定的交换机的模式type=ExchangeTypes.DIRECT)))@ComponentpublicclassConsumer01{// @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值@RabbitHandlerpublicvoidmessagerevice(Stringmessage){// 此处省略发邮件的逻辑System.out.println("c1-------------->"+message);}}

消费方c2

// bindings用来确定队列和交换机绑定关系@RabbitListener(bindings=@QueueBinding(//指定获取信息的队列value=@Queue(value=RabbitConfig.DEAD_QUEUE,autoDelete="false"),// 指定交换机exchange=@Exchange(value=RabbitConfig.DEAD_EXCHANGE,// 这里是确定的交换机的模式type=ExchangeTypes.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

队列达到最大长度代码演示

生产者取消发送消息的ttl

@ComponentpublicclassProducer{@AutowiredprivateRabbitTemplaterabbitTemplate;// 2: 路由keyprivateStringrouteKey="zhangsan";publicvoidmakeOrder(){// 发送消息给所有客户for(inti=1;i<=10;i++){Stringcontent="消息"+i;System.out.println("生产者发送消息:"+i)r abbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE,routeKey,content,null);}}}

设置普通队列的最大长度

//声明队列@Bean("normalQueueDeclared")publicQueuenormalQueueDeclared(){//正常队列绑定死信队列信息Map<String,Object>params=newHashMap<>();//正常队列设置死信交换机 参数 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(bindings=@QueueBinding(//指定获取信息的队列value=@Queue(value=RabbitConfig.NORMAL_QUEUE,autoDelete="false"),// 指定交换机exchange=@Exchange(value=RabbitConfig.NORMAL_EXCHANGE,// 这里是确定的交换机的模式type=ExchangeTypes.DIRECT)))@ComponentpublicclassConsumer01{// @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值@RabbitHandlerpublicvoidmessagerevice(Channelchannel,Messagemessage){Stringcontent=newString(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拒绝了一个消息，所以该消息被路由到了死信队列中。

死信队列的应用场景

一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息，这样比手工恢复数据要好太多了。

总结

死信队列其实并没有什么特别，不过是绑定在死信交换机上的普通队列，而死信交换机也只是一个普通的交换机，不过是用来专门处理死信的交换机。

死信队列之延时队列

延时队列链接