RabbitMQ可靠性投递与高可用架构

一，可靠性投递

使用 RabbitMQ 实现异步通信的时候，消息丢了怎么办，消息重复消费怎么办？

在 RabbitMQ 里面提供了很多保证消息可靠投递的机制，这个也是 RabbitMQ 的一个特性。

要明确一个问题，因为效率与可靠性是无法兼得的，如果要保证每一个环节都成功，势必会对消息的收发效率造成影响。所以如果是一些业务实时一致性要求不是特别高的场合，可以牺牲一些可靠性来换取效率。

比如发送通知或者记录日志的这种场景，如果用户没有收到通知，不会造成业务影响，只要再次发送就可以了。

在我们使用 RabbitMQ 收发消息的时候，有几个主要环节

①代表消息从生产者发送到 Broker

生产者把消息发到 Broker 之后，怎么知道自己的消息有没有被 Broker 成功接收？

②代表消息从 Exchange 路由到 Queue

Exchange 是一个绑定列表，如果消息没有办法路由到正确的队列，会发生什么事情？应该怎么处理？

③代表消息在 Queue 中存储

队列是一个独立运行的服务，有自己的数据库（Mnesia），它是真正用来存储消息的。如果还没有消费者来消费，那么消息要一直存储在队列里面。如果队列出了问题，消息肯定会丢失。怎么保证消息在队列稳定地存储呢？

④代表消费者订阅 Queue 并消费消息

队列的特性是什么？FIFO。队列里面的消息是一条一条的投递的，也就是说，只有上一条消息被消费者接收以后，才能把这一条消息从数据库删掉，继续投递下一条消息。那么问题来了，Broker 怎么知道消费者已经接收了消息呢？

1，消息发送到rabbitMQ服务器

第一个环节是生产者发送消息到 Broker。可能因为网络或者 Broker 的问题导致消息发送失败，生产者不能确定 Broker 有没有正确的接收。

在 RabbitMQ 里面提供了两种机制服务端确认机制，也就是在生产者发送消息给RabbitMQ 的服务端的时候，服务端会通过某种方式返回一个应答，只要生产者收到了这个应答，就知道消息发送成功了。

第一种是 Transaction（事务）模式，第二种 Confirm（确认）模式。

Transaction模式

事务模式怎么使用呢？

我们通过一个 channel.txSelect()的方法把信道设置成事务模式，然后就可以发布消息给 RabbitMQ 了，如果 channel.txCommit();的方法调用成功，就说明事务提交成功，则消息一定到达了 RabbitMQ 中。

如果在事务提交执行之前由于 RabbitMQ 异常崩溃或者其他原因抛出异常，这个时候我们便可以将其捕获，进而通过执行 channel.txRollback()方法来实现事务回滚。

在事务模式里面，只有收到了服务端的 Commit-OK 的指令，才能提交成功。所以可以解决生产者和服务端确认的问题。但是事务模式有一个缺点，它是阻塞的，一条消息没有发送完毕，不能发送下一条消息，它会榨干 RabbitMQ 服务器的性能。所以不建议大家在生产环境使用。

Spring Boot 中的设置

rabbitTemplate.setChannelTransacted(true);

/*** @author yhd* @createtime 2021/1/24 23:29* rabbitmq事务*/
@Component
public class MqTx {private static final String EXCHANGE = "exchange.tx";private static final String QUEUE = "queue.tx";private static final String ROUTING_KEY = "routing.tx";@Resourceprivate RabbitTemplate rabbitTemplate;public boolean sendMessage() {rabbitTemplate.setChannelTransacted(true);rabbitTemplate.setConfirmCallback((correlationData, flag, cause) -> {if (flag) {System.out.println("发送成功！");} else {System.out.println("发送失败" + cause);}});return true;}@RabbitListener(bindings =@QueueBinding(value = @Queue(value = QUEUE, autoDelete = "false", durable = "true"),exchange = @Exchange(value = EXCHANGE, autoDelete = "true", durable = "true"),key = {ROUTING_KEY}))public void receiveMessage(String msg, Message message, Channel channel) {try {channel.txSelect();System.out.println("msg = " + msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);channel.txCommit();} catch (IOException e) {try {channel.txRollback();} catch (IOException ioException) {ioException.printStackTrace();}}}
}

那么有没有其他可以保证消息被 Broker 接收，但是又不大量消耗性能的方式呢？这就是第二种模式，叫做确认（Confirm）模式。

Confirm （确认）模式

确认模式有三种，一种是普通确认模式。

在生产者这边通过调用 channel.confirmSelect()方法将信道设置为 Confirm 模式，然后发送消息。一旦消息被投递到所有匹配的队列之后，RabbitMQ 就会发送一个确认（Basic.Ack）给生产者，也就是调用 channel.waitForConfirms()返回 true，这样生产者就知道消息被服务端接收了。

这种发送 1 条确认 1 条的方式消息还不是太高，所以我们还有一种批量确认的方式。批量确认，就是在开启 Confirm 模式后，先发送一批消息。只要channel.waitForConfirmsOrDie();方法没有抛出异常，就代表消息都被服务端接收了。

批量确认的方式比单条确认的方式效率要高，但是也有两个问题，第一个就是批量的数量的确定。对于不同的业务，到底发送多少条消息确认一次？数量太少，效率提升不上去。数量多的话，又会带来另一个问题，比如我们发 1000 条消息才确认一次，如果前面 999 条消息都被服务端接收了，如果第 1000 条消息被拒绝了，那么前面所有的消息都要重发。

有没有一种方式，可以一边发送一边确认的呢？这个就是异步确认模式。

异步确认模式需要添加一个 ConfirmListener，并且用一个 SortedSet 来维护没有被确认的消息。

Confirm 模式是在 Channel 上开启的，因为 RabbitTemplate 对 Channel 进行了封装，叫做 ConfimrCallback。

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (!ack) {System.out.println("发送消息失败：" + cause);throw new RuntimeException("发送异常：" + cause);}});

2，消息从交换机路由到队列

第二个环节就是消息从交换机路由到队列。在什么情况下，消息会无法路由到正确的队列？可能因为路由键错误，或者队列不存在。

我们有两种方式处理无法路由的消息，一种就是让服务端重发给生产者，一种是让交换机路由到另一个备份的交换机。

消息回发的方式：使用 mandatory 参数和 ReturnListener（在 Spring AMQP 中是ReturnCallback）。

        rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnCallback((Message message,int replyCode,String replyText,String exchange,String routingKey) -> {// 反序列化对象输出log.info("消息主体: {}", new String(message.getBody()));log.info("应答码: {}", replyCode);log.info("描述：{}", replyText);log.info("消息使用的交换器 exchange : {}", exchange);log.info("消息使用的路由键 routing : {}", routingKey);});

消息路由到备份交换机的方式：在创建交换机的时候，从属性中指定备份交换机。

    private static final String EXCHANGE_NAME = "amqp.yhd.exchange";private static final String EXCHANGE_NAME_COPY = "amqp.yhd.exchange.copy";private static final String QUEUE_NAME = "amqp.yhd.queue";private static final String ROUTING_KEY = "amqp.admin";/*** AmqpAdmin** @param factory* @return*/@Beanpublic AmqpAdmin amqpAdmin(ConnectionFactory factory) {RabbitAdmin admin = new RabbitAdmin(factory);//给交换机指定备份交换机Map<String,Object> arguments = new HashMap();arguments.put("alternate-exchange",EXCHANGE_NAME_COPY); //声明一个交换机 交换机名  是否持久化  是否自动删除admin.declareExchange(new DirectExchange(EXCHANGE_NAME, true, false,arguments));//队列名 持久化 是否批处理  自动删除admin.declareQueue(new Queue(QUEUE_NAME, true, false, false));//声明一个绑定 队列名 ，绑定类型，交换机名，路由键 参数admin.declareBinding(new Binding(QUEUE_NAME, Binding.DestinationType.QUEUE, EXCHANGE_NAME, ROUTING_KEY, null));return admin;}

队列可以指定死信交换机；交换机可以指定备份交换机

3，消息在队列存储

第三个环节是消息在队列存储，如果没有消费者的话，队列一直存在在数据库中。

如果 RabbitMQ 的服务或者硬件发生故障，比如系统宕机、重启、关闭等等，可能会导致内存中的消息丢失，所以我们要把消息本身和元数据（队列、交换机、绑定）都保存到磁盘。

解决方案

队列持久化+交换机持久化+消息持久化

@Slf4j
@Component
public class MQProducerAckTest {@Autowiredprivate RabbitTemplate rabbitTemplate;private static final String EXCHANGE = "exchange.confirm";private static final String QUEUE = "queue.confirm";private static final String ROUTING_KEY = "routing.confirm";@Beanpublic AmqpAdmin amqpAdmin(ConnectionFactory factory) {RabbitAdmin admin = new RabbitAdmin(factory);//声明一个交换机 交换机名  是否持久化  是否自动删除admin.declareExchange(new DirectExchange(EXCHANGE, true, false, null));//队列名 持久化 是否批处理  自动删除admin.declareQueue(new org.springframework.amqp.core.Queue(QUEUE, true, false, false));//声明一个绑定 队列名 ，绑定类型，交换机名，路由键 参数admin.declareBinding(new Binding(QUEUE, Binding.DestinationType.QUEUE, EXCHANGE, ROUTING_KEY, null));return admin;}/*** 发送消息** @param exchange   交换机* @param routingKey 路由键* @param message    消息*/public boolean sendMessage(String exchange, String routingKey, String message) {MessageProperties messageProperties = new MessageProperties();messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message msg = new Message(message.getBytes(), messageProperties);rabbitTemplate.convertAndSend(exchange, routingKey, msg);return true;}@SneakyThrows@RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE, autoDelete = "false", durable = "true"),exchange = @Exchange(value = EXCHANGE, autoDelete = "true", durable = "true"),key = {ROUTING_KEY}))public void process(Message message, Channel channel) {log.info("RabbitListener:{}", new String(message.getBody()));channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}
}

集群

如果只有一个 RabbitMQ 的节点，即使交换机、队列、消息做了持久化，如果服务崩溃或者硬件发生故障，RabbitMQ 的服务一样是不可用的，所以为了提高 MQ 服务的可用性，保障消息的传输，我们需要有多个 RabbitMQ 的节点。

4，消息投递到消费者

如果消费者收到消息后没来得及处理即发生异常，或者处理过程中发生异常，会导致④失败。服务端应该以某种方式得知消费者对消息的接收情况，并决定是否重新投递这条消息给其他消费者。

RabbitMQ 提供了消费者的消息确认机制（message acknowledgement），消费者可以自动或者手动地发送 ACK 给服务端。

没有收到 ACK 的消息，消费者断开连接后，RabbitMQ 会把这条消息发送给其他消费者。如果没有其他消费者，消费者重启后会重新消费这条消息，重复执行业务逻辑。

消费者在订阅队列时，可以指定autoAck参数，当autoAck等于false时，RabbitMQ会等待消费者显式地回复确认信号后才从队列中移去消息。

如何设置手动 ACK？

SimpleRabbitListenerContainer 或者 SimpleRabbitListenerContainerFactory

factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

application.properties

spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual

注意这三个值的区别：NONE：自动 ACK，MANUAL：手动 ACK，AUTO：如果方法未抛出异常，则发送 ack。

当抛出 AmqpRejectAndDontRequeueException 异常的时候，则消息会被拒绝，且不重新入队。当抛出 ImmediateAcknowledgeAmqpException 异常，则消费者会发送 ACK。其他的异常，则消息会被拒绝，且 requeue = true 会重新入队。

在 Spring Boot 中，消费者又怎么调用 ACK，或者说怎么获得 Channel 参数呢？

    @SneakyThrows@RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE, autoDelete = "false", durable = "true"),exchange = @Exchange(value = EXCHANGE, autoDelete = "true", durable = "true"),key = {ROUTING_KEY}))public void process(Message message, Channel channel) {log.info("RabbitListener:{}", new String(message.getBody()));channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}

如果消息无法处理或者消费失败，也有两种拒绝的方式，Basic.Reject()拒绝单条，Basic.Nack()批量拒绝。如果 requeue 参数设置为 true，可以把这条消息重新存入队列，以便发给下一个消费者（当然，只有一个消费者的时候，这种方式可能会出现无限循环重复消费的情况。可以投递到新的队列中，或者只打印异常日志）。

服务端收到了 ACK 或者 NACK，生产者会知道吗？即使消费者没有接收到消息，或者消费时出现异常，生产者也是完全不知情的。

5，消费者回调

1）调用生产者 API

2）发送响应消息给生产者

6，补偿机制

如果生产者的 API 就是没有被调用，也没有收到消费者的响应消息，怎么办？

可能是消费者处理时间太长或者网络超时。

生产者与消费者之间应该约定一个超时时间，比如 5 分钟，对于超出这个时间没有得到响应的消息，可以设置一个定时重发的机制，但要发送间隔和控制次数，比如每隔 2分钟发送一次，最多重发 3 次，否则会造成消息堆积。

重发可以通过消息落库+定时任务来实现。

7，消息幂等性

如果消费者每一次接收生产者的消息都成功了，只是在响应或者调用 API 的时候出了问题，会不会出现消息的重复处理？

为了避免相同消息的重复处理，必须要采取一定的措施。RabbitMQ 服务端是没有这种控制的（同一批的消息有个递增的 DeliveryTag），它不知道你是不是就要把一条消息发送两次，只能在消费端控制。

如何避免消息的重复消费？

1、生产者的问题，环节①重复发送消息，比如在开启了 Confirm 模式但未收到确认，消费者重复投递。

2、环节④出了问题，由于消费者未发送 ACK 或者其他原因，消息重复投递。

3、生产者代码或者网络问题。

对于重复发送的消息，可以对每一条消息生成一个唯一的业务 ID，通过日志或者消息落库来做重复控制。

8，最终一致性

如果确实是消费者宕机了，或者代码出现了 BUG 导致无法正常消费，在我们尝试多次重发以后，消息最终也没有得到处理，怎么办？

手动处理

9，消息的顺序性

消息的顺序性指的是消费者消费消息的顺序跟生产者生产消息的顺序是一致的。

比如：1、发表微博；2、发表评论；3、删除微博。顺序不能颠倒。

在 RabbitMQ 中，一个队列有多个消费者时，由于不同的消费者消费消息的速度是不一样的，顺序无法保证。只有一个队列仅有一个消费者的情况才能保证顺序消费（不同的业务消息发送到不同的专用的队列）。

二，集群与高可用

1，为什么要做集群

集群主要用于实现高可用与负载均衡。

高可用：如果集群中的某些 MQ 服务器不可用，客户端还可以连接到其他 MQ 服务器。

负载均衡：在高并发的场景下，单台 MQ 服务器能处理的消息有限，可以分发给多台 MQ 服务器。

RabbitMQ 有两种集群模式：普通集群模式和镜像队列模式。

2，RabbitMQ 如何支持集群

应用做集群，需要面对数据同步和通信的问题。因为 Erlang 天生具备分布式的特性，所以 RabbitMQ 天然支持集群，不需要通过引入 ZK 或者数据库来实现数据同步。

RabbitMQ 通过/var/lib/rabbitmq/.erlang.cookie 来验证身份，需要在所有节点上保持一致。

3，rabbitMQ的节点类型

集群有两种节点类型，一种是磁盘节点（Disc Node），一种是内存节点（RAMNode）。

磁盘节点：将元数据（包括队列名字属性、交换机的类型名字属性、绑定、vhost）放在磁盘中。

内存节点：将元数据放在内存中。

PS：内存节点会将磁盘节点的地址存放在磁盘（不然重启后就没有办法同步数据了）。如果是持久化的消息，会同时存放在内存和磁盘。

集群中至少需要一个磁盘节点用来持久化元数据，否则全部内存节点崩溃时，就无从同步元数据。未指定类型的情况下，默认为磁盘节点。

我们一般把应用连接到内存节点（读写快），磁盘节点用来备份。

集群通过 25672 端口两两通信，需要开放防火墙的端口。

需要注意的是，RabbitMQ 集群无法搭建在广域网上

集群的配置步骤

1、配置 hosts

2、同步 erlang.cookie

3、加入集群（join cluster）

4，普通集群

普通集群模式下，不同的节点之间只会相互同步元数据。

为什么不直接把队列的内容（消息）在所有节点上复制一份？

主要是出于存储和同步数据的网络开销的考虑，如果所有节点都存储相同的数据，就无法达到线性地增加性能和存储容量的目的（堆机器）。

假如生产者连接的是节点 3，要将消息通过交换机 A 路由到队列 1，最终消息还是会转发到节点 1 上存储，因为队列 1 的内容只在节点 1 上。

同理，如果消费者连接是节点 2，要从队列 1 上拉取消息，消息会从节点 1 转发到节点 2。其它节点起到一个路由的作用，类似于指针。

普通集群模式不能保证队列的高可用性，因为队列内容不会复制。如果节点失效将导致相关队列不可用，因此我们需要第二种集群模式。

5，镜像集群

第二种集群模式叫做镜像队列。

镜像队列模式下，消息内容会在镜像节点间同步，可用性更高。不过也有一定的副作用，系统性能会降低，节点过多的情况下同步的代价比较大。

6，高可用

集群搭建成功后，如果有多个内存节点，那么生产者和消费者应该连接到哪个内存节点？如果在我们的代码中根据一定的策略来选择要使用的服务器，那每个地方都要修改，客户端的代码就会出现很多的重复，修改起来也比较麻烦。

所以需要一个负载均衡的组件（例如 HAProxy，LVS，Nignx），由负载的组件来做路由。这个时候，只需要连接到负载组件的 IP 地址就可以了。

负载分为四层负载和七层负载。

四层负载：工作在 OSI 模型的第四层，即传输层（TCP 位于第四层），它是根据 IP端口进行转发（LVS 支持四层负载）。RabbitMQ 是 TCP 的 5672 端口。

（修改报文中目标地址和原地址）

七层负载：工作在第七层，应用层（HTTP 位于第七层）。可以根据请求资源类型分配到后端服务器（Nginx 支持七层负载；HAProxy 支持四层和七层负载）。

（处理请求，代理至服务器）

但是，如果这个负载的组件也挂了呢？

我们应该需要这样一个组件

1、它本身有路由（负载）功能，可以监控集群中节点的状态（比如监控HAProxy），如果某个节点出现异常或者发生故障，就把它剔除掉。

2、为了提高可用性，它也可以部署多个服务，但是只有一个自动选举出来的 MASTER 服务器（叫做主路由器），通过广播心跳消息实现。

3、 MASTER 服务器对外提供一个虚拟 IP，提供各种网络功能。也就是谁抢占到 VIP，就由谁对外提供网络服务。应用端只需要连接到这一个 IP 就行了。

这个协议叫做 VRRP 协议（虚拟路由冗余协议 Virtual Router RedundancyProtocol），这个组件就是 Keepalived，它具有 Load Balance 和 High Availability的功能。

三，面试经验总结

1，到底在消费者创建还是在生产者创建？

如果A项目和B项目有相互发送和接收消息，应该创建几个vhost，几个Exchange？

交换机和队列，实际上是作为资源，由运维管理员创建的。

2，信息落库+ 定时任务

将需要发送的消息保存在数据库中，可以实现消息的可追溯和重复控制，需要配合定时任务来实现。

1）将需要发送的消息登记在消息表中。

2）定时任务一分钟或半分钟扫描一次，将未发送的消息发送到 MQ 服务器，并且修改状态为已发送。

3）如果需要重发消息，将指定消息的状态修改为未发送即可。

副作用：降低效率，浪费存储空间。

3，日志追踪

RabbitMQ 可以通过 Firehose 功能来记录消息流入流出的情况，用于调试，排错。

它是通过创建一个 TOPIC 类型的交换机（amq.rabbitmq.trace），把生产者发送给Broker 的消息或者 Broker 发送给消费者的消息发到这个默认的交换机上面来实现的。

另外 RabbitMQ 也提供了一个 Firehose 的 GUI 版本，就是 Tracing 插件。

启用 Tracing 插件后管理界面右侧选项卡会多一个 Tracing，可以添加相应的策略。

RabbitMQ 还提供了其他的插件来增强功能。

4，如何减少连接数

在发送大批量消息的情况下，创建和释放连接依然有不小的开销。我们可以跟接收方约定批量消息的格式，比如支持 JSON 数组的格式，通过合并消息内容，可以减少生产者/消费者与 Broker 的连接。

比如：活动过后，要全范围下线产品，通过 Excel 导入模板，通常有几万到几十万条解绑数据，合并发送的效率更高。

建议单条消息不要超过 4M（4096KB），一次发送的消息数需要合理地控制。

5，无法被路由的消息，去了哪里？

直接丢弃。可用备份交换机（alternate-exchange）接收。

6，大量消息堆积怎么办？

1）重启（不是开玩笑的）

2）多创建几个消费者同时消费

3）直接清空队列，重发消息

7，设计一个 MQ，你的思路是什么？

存储与转发。

存储：内存：用什么数据结构？

磁盘：文件系统？数据库？

通信：通信协议（TCP HTTP AMQP ）？一对一？一对多？一对多

推模式？拉模式？后者

其他特性……