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 的消息，值为false

2.否定确认

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:admin@47.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:admin@47.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 方法，如果消息成功发送到 Broker，ack 为 true

@Bean 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:admin@47.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:admin@47.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: manual

@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("业务处理完成"); 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 / Nack），requeue = false

2.消息过期

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:admin@47.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:admin@47.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而是继续接收消息