RabbitMQ 高级特性

消息确认机制生产者发送消息之后到达消费者之后会有几种情况消息处理成功或者处理失败RabbitMQ 向消费者发送消息之后就会自动删除这条消息要是消费失败了消息就丢失了为了保证消息从队列可靠地到达消费者RabbitMQ 提供了消费确认机制消费者在订阅队列时可以指定 aotoAck消息确认机制分为两种自动确认当 autoAck 等于 true时RabbitMQ 会自动把发送出去的消息置为确认然后从内存中删除不管消费者有没有真正消费这条消息手动确认当 autoAck 等于 false 时RabbitMQ 会等待消费者显式地调用Basic.Ack命令回复确认信号才从内存中移除消息// 消费消息 DefaultConsumer consumer new DefaultConsumer(channel){ // 从队列中接收到消息执行的方法 Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收到消息 new String(body)); } }; channel.basicConsume(Constants.WORK_QUEUE,true,consumer);当 autoAck 参数设置为 false,对于 RabbitMQ 服务端而言队列中的消息分为了两个部分:1.等待投递给消费者消息2.已经投递给消费者但是还没有收到消费者确认信号的信息如果 RabbitMQ 一直没有收到消费者的确认信号并且消费此消息的消费者已经断开连接则 RabbitMQ 会安排消息重新进入队列等待投递给下一个消费者当然也有可能是原来的哪个消费者Ready等待投递给消费者的消费数Unacked已经投递给消费者但是未收到消费者确认信号的消费数手动确认方法消费者在收到消息之后可以选择确认也可以直接拒绝或者跳过RabbitMQ 提供了三种不同的确认应答的方式消费者客户端可以调用与其对应的 channel 的相关方法1.肯定确认Channel.basicAck(long deliveryTag, boolean multiple)RabbitMQ 已经知道该消息并且成功的处理消息可以将其丢弃了deliverTag消息唯一标识deliverTag 是每个通道独立维护的在每个通道上都是唯一的当消费者确认一条消息时必须使用对应的通道上进行确认multiple是否批量确认为了减少网络流量可以对一系列连续的 deliverTag 进行批量确认值为 true 则会一次性 ack 所有小于或等于当前 deliverTag 的消息值为false2.否定确认Channel.basicReject(long deliveryTag, boolean requeue)RabbitMQ 2.0 版本开始引入 Basic.Reject 这个命令消费者客户端可以调用 channel.basicReject 方法来告诉 RabbitMQ 拒绝这个消息requeue在消息被拒绝后 如果设置为 true则 RabbitMQ 会重新把这条消息存入队列以便可以发送给下一个订阅的消费者如果设置为 false则不会重新入队列就这样不要了3.否定确认Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)Basic.Reject 命令一次只能拒绝一条消息如果想要批量拒绝消息则可以使用 Basic.Nack 这个命令消费者客户端可以调用 channel.basicNack 方法来实现multiple 设置为 true 表示拒绝deliverTag编号之前所有未被当前消费者确认的消息Spring-AMQP 对消息确认机制提供了三种策略1.AcknowledgeMode.NONE消息一旦投递给消费者不管消费者是否成功处理了消息RabbitMQ 就会自动确认消息从RabbitMQ 队列中移除消息如果消费者处理消息失败消息可能会丢失配置确认机制spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin /extension listener: simple: acknowledge-mode: none #??????声明交换机和队列// 消息确认 Bean(ackQueue) public Queue ackQueue(){ return QueueBuilder.durable(Constants.ACK_QUEUE).build(); } Bean(directExchange) public DirectExchange directExchange(){ return ExchangeBuilder.directExchange(Constants.ACK_EXCHANGE).build(); } /* Bean(ackBinding) public Binding ackBinding(Exchange directExchange,Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with(ack).noargs(); }*/ Bean(ackBinding) public Binding ackBinding(Qualifier(directExchange) DirectExchange directExchange,Qualifier(ackQueue) Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with(ack); }发送消息RestController RequestMapping(/producer) public class ProductController { Autowired private RabbitTemplate rabbitTemplate; RequestMapping(/ack) public String ack(){ rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,ack,consumer ack mode test); return 消息发送成功; } }消费者Component public class AckListener { RabbitListener(queues Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); // 消费者逻辑 System.out.printf(接收到消息: %s,deliverTag: %d \n,new String(message.getBody(),UTF-8),message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 System.out.println(业务逻辑处理); int num 3 / 0; System.out.println(业务处理完成); // 肯定确认 channel.basicAck(deliveryTag,false); } }可以看到消费者处理失败消息在RabbitMQ被移除了2. AcknowledgeMode.AUTO(默认)消费者在消息处理成功时会自动确认消息但如果处理过程中抛出了异常则不会确认消息配置信息spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension listener: simple: #acknowledge-mode: none #?????? acknowledge-mode: auto #??????发送消息日志上当消费者发生异常RabbitMQ就会一直重发由于异常多次重试还是失败消息没被确认无法nack一直处于 unacked 状态导致消息挤压3. AcknowledgeMode.MANUAL手动确认模式下消费者必须在成功处理后显示调用 basicAck 来确认消息如果消息未被确认RabbitMQ 会认为消息未被成功处理并且会在消费者可用时重新投递该消息这种模式提高了消息处理的可靠性因为即使消费者处理消息后失败消息也不会丢失而是可以被重新处理配置spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension listener: simple: #acknowledge-mode: none #?????? #acknowledge-mode: auto #?????? acknowledge-mode: manual #??????发送消息Component public class AckListener { RabbitListener(queues Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 消费者逻辑 System.out.printf(接收到消息: %s,deliverTag: %d \n,new String(message.getBody(),UTF-8),message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 System.out.println(业务逻辑处理); int num 3 / 0; System.out.println(业务处理完成); // 肯定确认 channel.basicAck(deliveryTag,false); } catch (Exception e) { // 发生异常进行否定确认 channel.basicNack(deliveryTag,false,false ); } } }basicNack 第三个参数时自动入队列我们此处并未开启持久性如何保证当 RabbitMQ 服务停掉之后生产者发送的消息不丢失默认情况下RabbitMQ 退出或由于某种原因崩溃时会忽视队列和消息RabbitMQ 的持久化分为三个部分交换器持久化队列持久化消息持久化交换机持久化交换机持久化是通过声明交换机时将 durable 参数设置为 true,相当于交换机的属性在服务器内部保存当 MQ 的服务器发生意外或关闭之后重启 RabbitMQ 时不需要重新去建立交换机交换机会自动建立ExchangeBuilder.topicExchange(Constant.ACK_EXCHANGE_NAME).durable(true).build()队列持久化队列的持久化是通过在声明队列时将 durable 参数设置为 true 实现的队列的持久化能保证该队列本身元数据在发生异常时不会丢失但是并不能保证内部所存储消息不会丢失QueueBuilder.durable(Constant.ACK_QUEUE).build();消息持久化message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);设置了队列和消息的持久化当 RabbitMQ 服务重启之后消息依旧存在如果只设置队列持久化重启之后消息会丢失如果只设置消息持久化重启之后队列消失消息也就丢失了RabbitMQ 默认情况下会将消息持久化除非队列被声明为非持久化但是如果将所有的消息都设置为持久化会严重影响 RabbitMQ 的性能写入磁盘的速度比内存慢的多得多我们需要在可靠性和吞吐量之间做一个权衡将交换机队列消息都设置为持久化之后是不能保证百分百数据不丢失的1.从消费者角度来说如果在订阅消费队列时将 autoAck 参数设置为 true那么当消费者接收到相关消息之后还没来得及处理就宕机了这也算数据丢失我们就需要设置为 手动确认2.在持久化消息正确存入 RabbitMQ 之后还需要有一段时间才能存入磁盘RabbitMQ 不会为每一条消息都进行同步存盘可能仅仅保存到操作系统缓存之中而不是物理磁盘中如果在这个中间状态下 RabbitMQ 服务节点发生了宕机重启等消息还没来得及放在物理磁盘中那么这些消息将会丢失解决方案1.引入 RabbitMQ的仲裁队列如果主节点在此特殊时间内挂掉可以自动切换到从节点这样保证了可用性除非整个集群都挂了2.还可以在发送端引入 事务机制 或者 发送方确认机制 来保证消息已经正确发送并存储 RabbitMQ中发送方确认在RabbitMQ中可以通过消息持久化来解决因服务器异常崩溃而导致的消息丢失但是当生产者将消息发送出去之后还没有正确的到达服务器比如 RabbitMQ 重启这样消息就真的没了服务器都没收到不能持久化解决方案1.通过事务机制2.通过 发送方确认机制实现我们来看RabbitMQ提供的两个控制消息的可靠性传递的模式confirm 确认模式Producer 发送消息的时候对发送端设置一个 ConfirmCallback 的监听无论消息是否到达 Exchange这个监听都会被执行如果 Exchange 成功收到消息Ack 为 true反之1.配置文件spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin /extension listener: simple: #acknowledge-mode: none #?????? #acknowledge-mode: auto #?????? acknowledge-mode: manual #?????? #prefetch : 1 publisher-confirm-type: correlated #消息发送确认声明交换机 和 队列//发送方确认 Bean(confirmQueue) public Queue confirmQueue(){ return QueueBuilder.durable(Constants.CONFIRM_QUEUE).build(); } Bean(confirmExchange) public DirectExchange confirmExchange(){ return ExchangeBuilder.directExchange(Constants.CONFIRM_EXCHANGE).build(); } Bean(confirmBinding) public Binding confirmBinding(Qualifier(confirmQueue) Queue queue,Qualifier(confirmExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(confirm).noargs(); }2.设置确认回调消息并发送消息无论消息确认成功还是失败都会调用 ConfirmCallback 的 confirm 方法如果消息成功发送到 Brokerack 为 trueBean public RabbitTemplate confirmRabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); // 设置回调方法 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(执行了confirm方法); if (ack){ System.out.printf(接收到消息消息ID%s \n,correlationData null ? null : correlationData.getId()); }else { System.out.printf(未接收到消息,消息ID: %s, cause : %s \n,correlationData null ? null : correlationData.getId(),cause); // 响应的业务处理比如重发 } } }); return rabbitTemplate; }RequestMapping(/producer) RestController public class ProducerController { Resource(name rabbitTemplate) private RabbitTemplate rabbitTemplate; Resource(name confirmRabbitTemplate) private RabbitTemplate confirmRabbitTemplate; RequestMapping(/confirm) public String confirm(){ CorrelationData correlationData new CorrelationData(1); confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE,confirm,confirm test .....,correlationData); return 消息发送成功; } }RabbitTemplate.ConfirmCallback 和 ConfirmListener 区别他们都是用来处理消息确认的机制但是他们属于不同的客户端库并且使用的场景和方式有所不同1.ConfirmListener 时 RabbitMQ Java Client 库中的接口它提供了两种方法: handleAck 和 handleNack,用于处理消息确认和否定确认的事件2.ConfirmCallback 时 Spring AMQP 框架中的一个接口用于简化与RabbitMQ交互的过程它只包含一个 confirm 方法用于处理消息确认的回调在 Spring Boot 应用中通常会使用 ConfirmCallback因为它与 Spring 框架的其他部分更加整合可以利用 Spring 的配置和依赖注入功能而在使用 RabbitMQ 和 Java Client 库时则可能会直接实现 ConfirmListener 接口更直接的与 RabbitMQ 的 Channel 交互3.测试失败会咋样改一下它找的交换机名称让他找不到交换机return 退回模式消息到达 Exchange 之后会根据路由规则匹配把消息放入 Queue 中Exchange 到 Queue 的过程如果一条消息无法被任何队列消费可以选择把消息退回给发送者消息退回给发送者时我们可以设置一个返回回调方法对消息进行处理1.配置文件spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension listener: simple: #acknowledge-mode: none #?????? #acknowledge-mode: auto #?????? acknowledge-mode: manual #?????? #prefetch : 1 publisher-confirm-type: correlated #消息发送确认2.设置返回回调逻辑并发送消息Bean public RabbitTemplate confirmRabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); // 消息被退回时回调方法 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { Override public void returnedMessage(ReturnedMessage returnedMessage) { System.out.println(消息退回 returnedMessage); } }); return rabbitTemplate; }RequestMapping(/returns) public String returns(){ CorrelationData correlationData new CorrelationData(5); confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE,confirm1,confirm test .....,correlationData); return 消息发送成功; }设置 RabbitTemplate 的 setMandatory 方法设置消息的 mandatory 属性为 true这个属性的作用是告诉 RabbitMQ如果一条消息无法被任何队列消费RabbitMQ 应该将消息返回给发送者此时 ReturnCallback 就会被触发3.测试常见面试题重点)如何保证RabbitMQ消息的可靠传输重试机制在消息传递过程中可能会遇到很多问题这些问题可能导致消息处理失败为了应对这种情况RabbitMQ 提供了重试机制允许消息在处理失败之后重新发送也可以设置从事次数以应对逻辑引起的错误自动确认配置文件spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension publisher-confirm-type: correlated #消息发送确认 listener: simple: #acknowledge-mode: none #?????? acknowledge-mode: auto #?????? #acknowledge-mode: manual #?????? #prefetch : 1 retry: enabled: true # 开启消费者失败重试 initial-interval: 5000ms # 初始失败等待5秒 max-attempts: 5 # 最大重试次数声明交换机和队列//重试机制 Bean(retryQueue) public Queue retryQueue(){ return QueueBuilder.durable(Constants.RETRY_QUEUE).build(); } Bean(retryExchange) public DirectExchange retryExchange(){ return ExchangeBuilder.directExchange(Constants.RETRY_EXCHANGE).build(); } Bean(retryBinding) public Binding retryBinding(Qualifier(retryQueue) Queue queue,Qualifier(retryExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(retry).noargs(); }发送消息RequestMapping(/retry) public String retry(){ rabbitTemplate.convertAndSend(Constants.RETRY_EXCHANGE,retry,retry test...........); return 消息发送成功; }消费消息Component public class RetryListener { RabbitListener(queues Constants.RETRY_QUEUE) public void handlerMessage(Message message) throws UnsupportedEncodingException { long deliveryTag message.getMessageProperties().getDeliveryTag(); System.out.printf([ Constants.RETRY_QUEUE ]接收到消息%s,deliveryTag: %s \n ,new String(message.getBody(),UTF-8),deliveryTag); System.out.println(业务处理完成); }尝试运行我们尝试出错会咋样RabbitListener(queues Constants.RETRY_QUEUE) public void handlerMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); System.out.printf([ Constants.RETRY_QUEUE ]接收到消息%s,deliveryTag: %s \n ,new String(message.getBody(),UTF-8),deliveryTag); try{ int num 3 / 0; System.out.println(业务处理完成); }catch (Exception e){ System.out.println(处理业务失败); } }手动确认acknowledge-mode: manualRabbitListener(queues Constants.RETRY_QUEUE) public void handlerMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); System.out.printf([ Constants.RETRY_QUEUE ]接收到消息%s,deliveryTag: %s \n ,new String(message.getBody(),UTF-8),deliveryTag); try{ int num 3 / 0; System.out.println(业务处理完成); channel.basicAck(deliveryTag,false); }catch (Exception e){ System.out.println(处理业务失败); channel.basicNack(deliveryTag,false,true); } }手动确认模式下重试次数的限制不会像在自动模式下那样直接生效因为是否重试以及何时重试更多取决于应用程序的逻辑和消费者的实现自动确认模式下RabbitMQ 会在消息被投递给消费者后自动确认消息如果消费者处理消息时抛出异常RabbitMQ 根据配置的重试次数自动将消息重新入队从而实现重试重试次数和重试间隔等参数直接在 RabbitMQ 配置中设定手动确认模式下消费者需要显示地对消息进行确认如果消费者在处理消息时遇到异常可以选择不确认消息可以重新入队重试的控制权在于应用程序本身应用程序可以通过自己的逻辑利用RabbitMQ 的高级特性来实现有效的重试策略重试机制需要注意的是1.自动确认模式下程序逻辑异常多次重试还是失败消息就会被自动确认那么消息就丢失了2.手动确认模式下程序逻辑异常多次重试消息依然处理失败无法被确认就会一直处于 unacked 的状态导致消息积压TTL过期时间RabbitMQ 可以对消息和队列设置 TTL当消息到达存活时间之后还没有被消费就会被自动清除设置消息 TTL设置消息 TTL 有两种方式1.设置队列 TTL队列中所有的消息都有相同的 过期时间2.对消息本身设置过期时间如果在设置了 TTL 的队列中设置存储设置了 TTL的消息则以较小的为准针对每条消息 TTL声明交换机 和 队列Bean(ttlQueue) public Queue ttlQueue(){ return QueueBuilder.durable(Constants.TTL_QUEUE).build(); } Bean(ttlExchange) public DirectExchange ttlExchange(){ return ExchangeBuilder.directExchange(Constants.TTL_EXCHANGE).build(); } Bean(ttlBinding) public Binding ttlBinding(Qualifier(ttlQueue) Queue queue,Qualifier(ttlExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ttl).noargs(); }发送消息RequestMapping(/ttl) public String ttl(){ rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,ttl,ttl test.........,message - { message.getMessageProperties().setExpiration(10000); return message; }); return 消息发送成功; }运行程序十秒之后设置队列 TTL// 设置 ttl 的队列 Bean(ttlQueue2) public Queue ttlQueue2(){ return QueueBuilder.durable(Constants.TTL_QUEUE2).ttl(20000).build(); } Bean(ttlBinding2) public Binding ttlBinding2(Qualifier(ttlQueue2) Queue queue,Qualifier(ttlExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ttl2).noargs(); }发送消息RequestMapping(/ttl2) public String ttl2(){ rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,ttl2,ttl2 test.........); return 消息发送成功; }运行程序20秒之后两者区别设置队列 TTL 属性的方法一旦消息过期就会被队列删除设置消息TTL的方法即使消息过期也不会马上从队列删除而是在即将投递到消费者之前进行判定两种方法处理方式为何不同因为设置队列过期时间看队列中已过期的消息肯定在队列头部RabbitMQ 只要定期从队头开始扫描是否有过期消息就可以了而设置消息TTL的方式每条消息的过期时间不同如果要删除所有过期消息需要扫描整个队列所以不如等到此消息被消费时再进行判定是否过期过期则删除死信队列死信 就是因为种种原因无法被消费的信息死信队列就是存储的死信当消息再一个队列中变成死信之后它能被重新发送到另一个交换器中这个交换器就是 DLX绑定DLX的队列就被称为死信队列消息变成死信有几种情况1.消息被拒绝Reject / Nackrequeue false2.消息过期3.队列达到最大长度示例声明队列和交换机//声明正常的交换机和队列 Bean(normalQueue) public Queue DLConfig(){ return QueueBuilder.durable(Constants.NROMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey(dlx) .maxLength(10L) .ttl(10000) .build(); } Bean(normalExchange) public DirectExchange normalExchange(){ return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).build(); } Bean(normalBinding) public Binding normalBinding(Qualifier(normalQueue) Queue queue,Qualifier(normalExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(normal).noargs(); } // 死信交换机和队列 Bean(dlQueue) public Queue dlQueue(){ return QueueBuilder.durable(Constants.DL_QUEUE).build(); } Bean(dlExchange) public DirectExchange dlExchange(){ return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).build(); } Bean(dlBinding) public Binding dlBinding(Qualifier(dlQueue) Queue queue,Qualifier(dlExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(dlx).noargs(); }此处我们不仅要声明死信队列和交换机还要生成正常的交换机和队列并保证他们要能生成出 死信我们这里选择了 TTL 队列发送消息RequestMapping(/dl) public String dl(){ rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,dl test.........); return 消息发送成功; }测试死信D持久化Lim设置了队列长度DLX死信交换机DLK死信 RountingKey十秒之后死信跑去了死信队列测试队列长度消息进入死信队列RequestMapping(/dl) public String dl(){ //测试队列长度 for (int i 0; i 20; i) { rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,dl test......... i); } return 消息发送成功; }记住我们在设置正常队列的时候设置的长度是 10测试消息拒收Component public class DLListener { RabbitListener(queues Constants.NROMAL_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 消费者逻辑 System.out.printf([normal.queue]接收到消息: %s,deliverTag: %d \n, new String(message.getBody(), UTF-8), message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 System.out.println(业务逻辑处理); int num 3 / 0; System.out.println(业务处理完成); // 肯定确认 channel.basicAck(deliveryTag, false); } catch (Exception e) { // 发生异常进行否定确认 channel.basicNack(deliveryTag, false, false);// 变成死信 } } RabbitListener(queues Constants.DL_QUEUE) public void handMessage1(Message message, Channel channel) throws Exception { System.out.printf([dl.queue]接收到消息: %s,deliverTag: %d \n, new String(message.getBody(), UTF-8), message.getMessageProperties().getDeliveryTag()); } }我们需要让消费者拒收正常队列的消息这样死信队列的消费者才能消费这些消息常见面试题重点死信的概念死信是消息队列中的一种特殊消息他指的是那些无法被正常消费或处理的消息死信的来源1.消息过期2.消息被拒绝3.队列满了死信队列应用场景在用户支付系统中用户支付订单之后支付系统会给订单系统返回当前订单的支付状态为了保证支付信息不丢失需要用到死信队列当消息消费异常时将消息投入到死信队列中由订单系统的其他消费者来监听这个队列并对数据进行处理消息重试将这个队列的死信发给之前的队列或者发给另一个队列进行重试消息丢弃延迟队列消息被发送之后并不想消费者立刻拿到消息而是等待特定事件之后消费者才能拿到这个消息RabbitMQ 本身没有这个功能但是可以使用 TTL 死信队列 的方式模拟出一个延迟队列比如说我们将要延迟的消息设置为过期时间放在正常队列之中等时间到了他自然就过期了就达到我们延迟的目的了然后这个消息就会被放到死信队列死信队列再发给消费者但是我们这样做的还是有缺陷的比如如果我们先发一个 10s 的信息再发 20s 的信息这样是 10s 先到死信然后被消费20s 则是后来者但是如果反过来就会出现一个问题队列检查过期时间是检查队头的如果我们先发一个 10000s 的消息再发一个 1s 的消息的话等到发现 1s 消息过期的时候已经是 10000s 之后了这很不合理。延迟队列插件具体的下载文件在 github过程我们不演示了验证插件x-delayed-message 是新加的基于插件延迟队列实现声明交换机和队列Configuration public class DelayConfig { Bean(delayQueue) public Queue delayQueue(){ return QueueBuilder.durable(Constants.DELAY_QUEUE).build(); } Bean(delayExchange) public Exchange delayExchange(){ return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build(); } Bean(delayBinding) public Binding delayBinding(Qualifier(delayQueue) Queue queue,Qualifier(delayExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(delay).noargs(); } }生产者RequestMapping(/delay) public String delay(){ rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,delay,delay test 30s.........,message - { message.getMessageProperties().setDelayLong(30000L); return message; }); rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,delay,delay test 10s.........,message - { message.getMessageProperties().setDelayLong(10000L); return message; }); return 消息发送成功; }消费者Component public class DelayListener { RabbitListener(queues Constants.DELAY_QUEUE) public void handMessage1(Message message, Channel channel) throws Exception { System.out.printf([delay.queue] %tc 接收到消息: %s \n,new Date(),new String(message.getBody(),UTF-8)); } }尝试运行通过控制台我们终于得到我们想要的结果了这个方法是支持乱序延迟的常见面试题重点介绍延迟队列是一个特殊的队列消息发送之后并不立即给消费者而是等待特定的时间才发送给消费者延迟队列的应用场景有1.订单在十分钟内没有支付自动取消2.用户注册成功之后3天后发调查问卷3.用户发起退款商家 24 小时后未处理则自动退款RabbitMQ 实现延迟队列有两种方法1.TTL 死信队列2.使用 RabbitMQ 官方的延迟插件二者对比1.基于死信队列实现优点灵活不需要插件缺点不支持消息乱序需要额外的逻辑来处理死信队列增加了系统复杂性2.基于插件实现延迟队列优点通过插件可以直接实现延迟插件拥有消息乱序的能力缺点需要以来特定的插件只适用于特定的版本事务RabbitMQ 是基于 AMQP 协议实现的该协议实现了事务机制因此 RabbitMQ 也支持事务机制Spring AMQP 也提供了事务相关操作RabbitMQ 事务允许开发者确保消息的发送和接收是原子性的要么全部成功要么全部失败配置事务管理器Bean(transTemplate) public RabbitTemplate transTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); rabbitTemplate.setChannelTransacted(true); return rabbitTemplate; } Bean public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory){ return new RabbitTransactionManager(connectionFactory); }声明队列//事务 Bean(transQueue) public Queue transQueue(){ return QueueBuilder.durable(Constants.TRANS_QUEUE).build(); }生产者我们尝试不加事务注解//Transactional RequestMapping(/trans) public String trans(){ System.out.println(trans test....); transTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test 1....); int a 5 / 0; transTemplate.convertAndSend(,Constants.TRANS_QUEUE,trans test 2....); return 消息发送成功; }加上注解呢消息分发RabbitMQ 队列拥有多个消费者时队列会把收到的消息分派给不同的消费者每条消息只会发送给订阅列表里的一个消费者这种方式非常适合扩展如果现在负载加重那么就需要创建更多的消费者来消费消息默认情况下RabbitMQ 是以轮询的方法进行分发的而不管消费者是否已经消费并确认了消息这种方式是不太合理的对于消费消息慢的消费者就会造成消息积压而消费消息快的可能出现空闲进而应用整体的吞吐量下降我们可以使用 channel.basicQos(intprefetchCount)⽅法来限制当前信道上的消费者所能保持的最大未确认消息的数量应用场景1.限流 2.非公平分发限流订单系统每秒最多处理 5000 请求正常情况下订单系统可以正常满足需求但是在秒杀时间点请求量瞬增每秒 1万 个请求如果这些请求全部通过MQ发送到订单系统会将订单系统压垮RabbitMQ 提供了限流机制可以控制消费者一次只拉取 N 个请求通过设置 prefetchCount 参数同时也必须要设置消息应答方式为手动应答prefetchCount控制消费者从队列中拉取消息的数量以此来实现流控制和负载均衡配置文件spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension #publisher-confirm-type: correlated #消息发送确认 listener: simple: #acknowledge-mode: none #?????? #acknowledge-mode: auto #?????? acknowledge-mode: manual #?????? prefetch : 5声明交换机和队列Configuration public class QosConfig { Bean(qosQueue) public Queue qosQueue(){ return QueueBuilder.durable(Constants.QOS_QUEUE).build(); } Bean(qosExchange) public Exchange qosExchange(){ return ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build(); } Bean(qosBinding) public Binding qosBinding(Qualifier(qosQueue) Queue queue,Qualifier(qosExchange) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(qos).noargs(); } }发送消息RequestMapping(/qos) public String qos(){ for (int i 0; i 20; i) { rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE,qos,qos test......... i); } return 消息发送成功; }消费者Component public class QosListener { RabbitListener(queues Constants.QOS_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 消费者逻辑 System.out.printf(接收到消息: %s,deliverTag: %d \n,new String(message.getBody(),UTF-8),message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 Thread.sleep(2000); // 肯定确认 //channel.basicAck(deliveryTag,false); } catch (Exception e) { // 发生异常进行否定确认 channel.basicNack(deliveryTag,false,false); } } }我们故意不让消费者消费消息来看它一次性能消费多少个消息负载均衡在有两个消费者的情况下一个消费者处理任务非常快另一个非常慢就会造成一个消费者一直很忙而另一个很闲RabbitMQ 只是在消息进入队列时分派消息不考虑消费者未确认消息的数量我们限制这俩消费者一次只能消费一个消息这样的话他俩一直处于很忙的状态spring: application: name: rabbit-extension-demo rabbitmq: addresses: amqp://admin:admin47.101.223.145:5672/extension #publisher-confirm-type: correlated #消息发送确认 listener: simple: #acknowledge-mode: none #?????? #acknowledge-mode: auto #?????? acknowledge-mode: manual #?????? prefetch : 1启动两个消费者Component public class QosListener { RabbitListener(queues Constants.QOS_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 消费者逻辑 System.out.printf(接收到消息: %s,deliverTag: %d \n,new String(message.getBody(),UTF-8),message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 Thread.sleep(2000); // 肯定确认 channel.basicAck(deliveryTag,false); } catch (Exception e) { // 发生异常进行否定确认 channel.basicNack(deliveryTag,false,false); } } RabbitListener(queues Constants.QOS_QUEUE) public void handMessage2(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 消费者逻辑 System.out.printf(222接收到消息: %s,deliverTag: %d \n,new String(message.getBody(),UTF-8),message.getMessageProperties().getDeliveryTag()); // 进行业务逻辑处理 Thread.sleep(1000); // 肯定确认 channel.basicAck(deliveryTag,false); } catch (Exception e) { // 发生异常进行否定确认 channel.basicNack(deliveryTag,false,false); } } }消费者2是比消费者1快的我们来看看测试结果可以看到即使消费者2已经完成了自己的任务但是它并没有等待消费者1而是继续接收消息