目录
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.代码实战
先写两个队列和分别自定义一个直连交换机 并且给交换机分别绑定一个队列
//死信,延迟队列@Beanpublic 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);}@Beanpublic DirectExchange ExchangeA(){return new DirectExchange("ExchangeA");}@Beanpublic Binding bindingA(){return BindingBuilder.bind(queueA()).to(ExchangeA()).with("aa");}@Beanpublic Queue queueB(){return new Queue("queueB");}@Beanpublic DirectExchange ExchangeB(){return new DirectExchange("ExchangeB");}@Beanpublic 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 {@RabbitHandlerpublic 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.132password: 123456port: 5672username: springvirtual-host: my_vhostlistener: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 {@RabbitHandlerpublic void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{log.warn("QA接收到:" + id);channel.basicAck(tag,true);}
}
在页面中刷新上面的方法,可以在idea中看到
如果想要手动否认,拒绝消息
@RabbitHandlerpublic 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
也可以拒绝该消息,消息会被丢弃,不会重回队列
@RabbitHandlerpublic 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