2.RabbitMQ常见高级特性

本文总结RabbitMQ常见的高级特性,希望能每次看到这篇文章都可以快速理清并理解RabbitMQ的高级特性.

消息传输大致流程图:

一.消息确认机制:包含两部分("生产者→Broker(RabbitMQ服务器)"和"Broker--->消费者"这一过程):

1.生产者→Broker:发送方确认机制

有两种:confrim模式

returns模式

2.Broker--->消费者:消费者 ACK 确认机制

有两种:①:自动确认(默认)(配置autoAck=true)--->针对消息可靠性要求不是很高的场景

②:手动确认(配置autoAck=false)--->消息可靠性要求较高的场景

👍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)

参数:<1><2>同上

<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方法中设置requeue=true参数,来将异常消息重新入队,从而实现重试的效果.

⭐自动确认的重试机制和手动确认的重新入队实现重试效果的区别:

<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.0

<3>放到 RabbitMQ 的plugins目录

<4>执行启用命令：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

<5>重启 RabbitMQ

<6>使用SpringBoot实现代码时,需要添加依赖

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

👍9.事务

RabbitMQ是基于AMQP协议实现的,由于AMQP协议实现了事务的机制,所以RabbitMQ也支持事务机制.RabbitMQ事务允许开发者确保消息的发送和接收是原子性的,要么全部成功,要么全部失败.

实现事务三步走(三条都不可少)

①:自定义RabbitMQTemplate

@Configuration public class RabbitTemplateConfig { @Bean("transRabbitTemplate") public RabbitTemplate transRabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); rabbitTemplate.setChannelTransacted(true); return rabbitTemplate; } }

②:生产者发送消息

③:自定义RabbitTemplate中定义RabbitTransactionMannager

@Configuration 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 num=5/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 num=5/0; transRabbitTemplate.convertAndSend("",Constants.TRANS_QUEUE,"trans test2..."); return "消息发送成功"; } }

👍10.消息分发

是什么:将消息分发给多个同时监听同一个队列的不同消费者

分发方式:

①:限流

②非公平分发(负载均衡)

代码实现:

<1>(全局)限流(本质:限制单个消费者持有的最大未确认消息数)

1`配置yml/properties:

spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://guest:guest@localhost:8889/extension listener: simple: acknowledge-mode: manual # 手动确认 prefetch: 1 # 限流核心：每次只处理1条消息

<2>负载均衡(本质:设置多个消费者监听同一个队列时,消息默认轮询分发)