目录
1.死信交换机介绍
2.TTL
3.延迟队列
4.消息堆积问题
5.惰性队列
6.代码实战
1.死信交换机介绍
当一个队列中信息满足下列情况之一时，可以成为死信（dead letter）
（1）消费者使用basic.reject（RejectAndDontRequeueRecoverer处理策略）或bastic.nack（ImmediateRequeueMessageRecoverer处理策略）消费失败，requeue（重新入队）设置参数为false；
（2）消息是一个过期消息，过期无人消费；
（3）要投递的队列消息堆积满了，最早的消息成为死信；
如果该队列配置了dead-letter-exchange属性，指定交换机，队列设置dead-letter-routing-key属性，设置死信交换机与死信队列的RoutingKey，那么队列中的死信就会投递到这个交换机中，而这个交换机成为死信交换机（Dead Letter Exchange，简称DLX）。
2.TTL
TTL，也就是Time-To-Live。如果队列中的消息TTL结束仍未消费则会成为死信，TTL超时分为两种情况：
（1）消息所在的队列设置了存活时间；
（2）消息本身设置了存活时间；
3.延迟队列
利用ttl结合死信交换机，实现了消息发出后，消费者延迟收到消息的结果，这种消息队列成为延迟队列（Delay Queue）模式。
延迟队列使用的场景：
（1）用户下单后超过15分钟未支付取消支付；
（2）预约20分钟会议，到时间自动通知参会人员；
订单的超时处理：
生产者生产一条1分钟后超时的订单消息到正常交换机exchange中，但一分钟后仍未消费，消息会被投递到死信交换机dlx-exchange中，并发送到私信队列中，死信队列dlx-queue的消费者拿到消息后，根据消息去查询订单的状态，如果仍然是为支付状态，将订单状态更新为超时状态
4.消息堆积问题
当生产者发送的消息的速度大于消费者消费的速度，就会导致队列中的消息堆积，直到达到消息的上限，最早接受的消息，可能会被丢弃，成为死信。
解决消费堆积的三种策略：
（1）增加消费者，增加消费速度
（2）在消费者内开启线程池加快消息处理速度
（3）扩大队列的容积
5.惰性队列
从RabbitMq的3.6.0版本开始增加了Lazy Queues的概念，也就是惰性队列。
惰性队列的特征如下：
（1）接受消息后直接存入磁盘而非内存；
（2）消费者消费消息才会中磁盘中将消息加载到内存；
（3）支持数百万的数据存储；
6.代码实战
先写两个队列和分别自定义一个直连交换机 并且给交换机分别绑定一个队列
//死信，延迟队列
@Bean
public Queue queueA(){
Map<String, Object> config = new HashMap<>();
//message在该队列queue的存活时间最大为20秒
config.put("x-message-ttl", 20000);
//x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
config.put("x-dead-letter-exchange", "ExchangeB");
//x-dead-letter-routing-key参数是给这个DLX指定路由键
config.put("x-dead-letter-routing-key", "bb");
return new Queue("queueA",true,false,false,config);
}
@Bean
public DirectExchange ExchangeA(){
return new DirectExchange("ExchangeA");
}
@Bean
public Binding bindingA(){
return BindingBuilder.bind(queueA()).to(ExchangeA()).with("aa");
}
@Bean
public Queue queueB(){
return new Queue("queueB");
}
@Bean
public DirectExchange ExchangeB(){
return new DirectExchange("ExchangeB");
}
@Bean
public Binding bindingB(){
return BindingBuilder.bind(queueB()).to(ExchangeB()).with("bb");
}在控制类给交换机发送一条消息
//死信交换机
@RequestMapping("/send5")
public String send5(){
template.convertAndSend("ExchangeA","aa","4502520");
return "😡";
}写一个测试类进行测试
package com.example.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues="queueB")
public class ReceiverQB {
@RabbitHandler
public void process(String id){
log.warn("QB接收到:" + id);
}
}
接着启动生产者和消费者，并在RabbitMQ中查看
注意：如果在启动消费者时报错，请先在页面上访问，再重新启动消费者
因为我们设置队列queue的存活时间最大为20秒，所以再隔20秒后消息会自动出来
确认消息（局部方法处理消息）
默认情况下消息消费者是自动 ack （确认）消息的，如果要手动 ack（确认）则需要修改确认模式为 manual
在消费者的yml文件中进行配置
server:
port: 9999
spring:
rabbitmq:
host: 192.168.169.132
password: 123456
port: 5672
username: spring
virtual-host: my_vhost
listener:
simple:
acknowledge-mode: manual接着写一个测试类
package com.example.consumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues="queueA")
public class ReceiverQA {
@RabbitHandler
public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
log.warn("QA接收到:" + id);
channel.basicAck(tag,true);
}
}
在页面中刷新上面的方法，可以在idea中看到
如果想要手动否认，拒绝消息
@RabbitHandler
public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
log.error("QA接收到:" + id);
channel.basicReject(tag,true);
Thread.sleep(1000);//延迟时间
}因为延迟时间只有1秒钟，而且需要重新进入队列，所以会一直循环，在循环到在我们上面设置的那个20秒钟后会进入QB
也可以拒绝该消息，消息会被丢弃，不会重回队列
@RabbitHandler
public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
log.error("QA接收到:" + id);
channel.basicReject(tag,false);
Thread.sleep(1000);//延迟时间
}这时候会直接进入QB
