2.RabbitMQ常见高级特性

本文总结RabbitMQ常见的高级特性,希望能每次看到这篇文章都可以快速理清并理解RabbitMQ的高级特性.消息传输大致流程图:一.消息确认机制:包含两部分(生产者→Broker(RabbitMQ服务器)和Broker---消费者这一过程):1.生产者→Broker:发送方确认机制有两种:confrim模式returns模式2.Broker---消费者:消费者 ACK 确认机制有两种:①:自动确认(默认)(配置autoAcktrue)---针对消息可靠性要求不是很高的场景②:手动确认(配置autoAckfalse)---消息可靠性要求较高的场景1.手动确认方法①:basicAck:消息处理成功,确认签收格式:channel.basicAck(long deliveryTag,boolean multiple)参数:1 deliveryTag:消息唯一标识,MQ用来区分消息2 multiple:{ true: 批量确认tag的所有未被确认的消息false: 只确认当前这条消息②:basicNack:消息处理失败,拒绝消息(支持批量拒绝)格式:channel.basicNack(long deliveryTag,boolean multiple,boolean requeue)参数:12同上3 requeue(是否重新入队):{true:消息重新入队false:直接丢弃消息注:由于消息传输可能成功也可能失败,所以建议basicAck basicNack 配套写③:basicReject:拒绝单挑消息(更轻量,和basicNack功能一样,但是只能拒绝单挑消息,不支持批量)格式:channel.basicReject(long deliveryTag,boolean requeue)参数:同上消息确认SpringBoot代码编写简单样例:Component public class Consumer { RabbitListener(queues test_queue) public void receive(String msg, Channel channel, Message message) { long tag message.getMessageProperties().getDeliveryTag(); try { // 1. 业务处理 System.out.println(消费 msg); // 2. 成功确认 channel.basicAck(tag, false); } catch (Exception e) { // 3. 失败拒绝 channel.basicNack(tag, false, true); } } }2.持久化(RabbitMQ可靠性保证机制之一):RabbitMQ持久化分{①:交换机持久化②:队列持久化③:消息持久化SpringBoot代码展示:1交换机持久化2队列持久化Configuration public class RabbitMQConfig { //持久化 Bean(presQueue) public Queue presQueue(){ //队列持久化 return QueueBuilder.durable(Constants.PRES_QUEUE).build(); //队列非持久化 //return QueueBuilder.nonDurable(Constants.PRES_QUEUE).build(); } Bean(presExchange) public DirectExchange presExchange(){ //交换机持久化 return ExchangeBuilder.directExchange(Constants.PRES_EXCHANGE).durable(true).build(); //交换机非持久化 //return ExchangeBuilder.directExchange(Constants.PRES_EXCHANGE).durable(false).build(); } //绑定关系 Bean(presBinding) public Binding presBinding(Qualifier(presQueue) Queue queue,Qualifier(presExchange) DirectExchange directExchange){ return BindingBuilder.bind(queue).to(directExchange).with(pres); } }3消息持久化RequestMapping(/producer) RestController public class ProducerController { //持久化 RequestMapping(/pres) public String pres() { Message message new Message(Presistent test....getBytes(), new MessageProperties()); //消息非持久化 //message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); //消息持久化 message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); System.out.println(message); rabbitTemplate.convertAndSend(Constants.PRES_EXCHANGE, pres, message); return 消息发送成功; } }⭐因为消息是存在队列中的,所以消息要持久化,则需要保证交换机持久化队列持久化消息持久化.3.发送方确认(publisher confirm)为避免消息在生产者发送后到达服务器之前丢失,解决方案有:a.通过事务机制实现(用的少)b.通过发送方确认机制实现(常用)------①:confirm模式②:returns模式二者可单独使用或者一起配合使用tips: confirm模式是应用在生产者到交换机,而returns模式应用在交换机到队列4.重试机制(需要yml/properties配置)注:重试机制需要再消息确认方式为自动时,才会生效,那么手动确认怎么实现重试机制这样的效果呢?手动确认其实可以通过在basicNack或basicReject方法中设置requeuetrue参数,来将异常消息重新入队,从而实现重试的效果.⭐自动确认的重试机制和手动确认的重新入队实现重试效果的区别:1自动确认的重试是本地反复重试2手动确认的重试是退回队列重发二者不是同一个机制配置代码:spring: rabbitmq: host: localhost port: 5672 username: guest password: guest listener: simple: acknowledge-mode: auto # 消息自动确认 retry: enabled: true # 开启消费者失败重试 max-attempts: 3 # 最大重试次数,这里是最多重试3次 initial-interval: 1000 # 第一次重试间隔1秒5.设置消息TTL(过期时间)TTL概念:是过期时间,消息的寿命,存活时间,当消息到达设置的过期时间还没有被消费掉,就会被消除.设置消息TTL方式有二:①:设置队列的TTl(存在该队列的所有消息的TTl与队列一致)②:设置消息TTL(消息可单独设置TTL)注:当队列和消息同时设置了同一TTL时,实际TTL取较小值二者区别:①:在设置队列TTL的方法时,一旦消息过期,就会从队列中删除②:设置消息TTL的方法时,即使消息过期,也不会马上从队列中删除,而是在该消息即将被消费者消费之前进行判定后再进行删除.与redis中的key的过期机制的惰性删除一样,6.死信队列(处理消息异常的情况)概念:死信:正常队列里无法被正常消费的消息死信队列:用来存死信的队列消息变成死信一般有以下几种情况:①:消息被消费者拒绝②:消息过期③:队列中的消息数量超出了队列设置的最大长度8.延迟队列由于RabbitMQ本身不直接支持延迟队列的实现,所以可以通过TTL死信队列的方式来间接实现延迟队列延迟队列通过TTL死信队列的实现方式:①设置队列的TTL死信队列----没问题,统一了队列中所有的消息的TTL②设置消息的TTL死信队列----可能有问题当后来的消息的过期时间早于先到的消息时,延迟功能可能会出问题,不生效.如:先发送30s的TTL消息,再发送10s的TTL消息,而10s的TTL消息会在30s的消息过期之后,才进行处理,这明显不合理.需要使用官方提供的延迟插件来专门解决这个问题,从而顺利实现延迟功能⭐引入延迟插件步骤:1需要在RabbitMQ服务端进行安装2下载 .ez插件文件要和 RabbitMQ 版本一致GitHub下载链接https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/v4.2.03放到 RabbitMQ 的plugins目录4执行启用命令rabbitmq-plugins enable rabbitmq_delayed_message_exchange5重启 RabbitMQ6使用SpringBoot实现代码时,需要添加依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency9.事务RabbitMQ是基于AMQP协议实现的,由于AMQP协议实现了事务的机制,所以RabbitMQ也支持事务机制.RabbitMQ事务允许开发者确保消息的发送和接收是原子性的,要么全部成功,要么全部失败.实现事务三步走(三条都不可少)①:自定义RabbitMQTemplateConfiguration public class RabbitTemplateConfig { Bean(transRabbitTemplate) public RabbitTemplate transRabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); rabbitTemplate.setChannelTransacted(true); return rabbitTemplate; } }②:生产者发送消息③:自定义RabbitTemplate中定义RabbitTransactionMannagerConfiguration public class RabbitTemplateConfig { Bean public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) { return new RabbitTransactionManager(connectionFactory); } }采用事务后,两条消息中间无异常---同时成功成功执行第一条消息后,在执行第二条消息之前有异常---同时失败,第一条成功执行的也认为失败RequestMapping(/producer) RestController public class ProducerController { Resource(name transRabbitTemplate) private RabbitTemplate transRabbitTemplate; //不采用事务的方式,第一条消息成功,第二条失败 RequestMapping(/trans) public String trans() { System.out.println(trans test...); rabbitTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test1...); int num5/0; rabbitTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test2...); return 消息发送成功; } //采用事务的方式,要么期间无异常同时成功,要么期间有异常同时失败 Transactional RequestMapping(/trans2) public String trans2() { System.out.println(trans test...); transRabbitTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test1...); //int num5/0; transRabbitTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test2...); return 消息发送成功; } }10.消息分发是什么:将消息分发给多个同时监听同一个队列的不同消费者分发方式:①:限流②非公平分发(负载均衡)代码实现:1(全局)限流(本质:限制单个消费者持有的最大未确认消息数)1配置yml/properties:spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://guest:guestlocalhost:8889/extension listener: simple: acknowledge-mode: manual # 手动确认 prefetch: 1 # 限流核心每次只处理1条消息2负载均衡(本质:设置多个消费者监听同一个队列时,消息默认轮询分发)