《rabbitmq 实战指南》- 读书笔记

快来打我* 2022-01-19 23:19 389阅读 0赞

rabbitmq的主要作用有：异步，解耦，削峰填谷

rabbitmq的生命周期包含连接、创建、生产、消费，关闭。

rabbitmq的消费模式有两种，push和pull，推模式采用basic.consume消费,拉模式则调用basic.get进行消费

rabbitmq提供了消息确认机制，目的是确保消息一定能够被消费者消费掉，当消费者订阅队列的时候，可以指定autoAck参数，为false的时候

rabbitmq会等待消费者显示回复确认消息后才从内存或磁盘中移除消息(移除的过程是先打上删除标记，之后删除)。当autoAck参数为true时，

则rabbitmq会把发送出去的消息置为确认，然后直接删除，不管消费者那边的情况。一直未确认的消息将一直等待在队列中，没有过期时间，判断

为确认消息是否重新投递给消费者的唯一依据就是该消息的消费者是否已经断开（断开则重新投递）。

消费者接收到消息后可以选择拒绝or接收，当选择拒绝时，如果requeue参数为true，则会重新存入队列投递给其他消费者，否则就直接删除消息。默认该参数为true

rabbitmq交换器类型有fanout,direct,topic,headers这四种，AMQP协议还提到了system和自定义两种

*  fanout:会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中去
 *  direct:会把消息路由到那些BindingKey和RoutingKey完全匹配的队列中去(即路由键要匹配)
 *  topic:与direct相似，不过它匹配规则有些不同，支持模糊匹配
 *  headers:不依赖路由键进行匹配消息，而是根据消息内容中的headers属性匹配，headers交换器性能很差，而且不实用,一般不用

第四章 rabbitmq进阶

mandatory参数告诉服务器至少该消息能路由到一个队列中，否则消息返回给生产者。immediate参数告诉服务器至少该消息关联的队列上有消费者，否则返还给生产者，不用投递到队列中等待消费者了（也就是设置为true）,rabbitmq 3.0取消了immediate参数支持，改为TTL和DLX替代

可以设置一个备份交换器来替代设置mandatory参数的复杂度，当消息不能正常路由到队列中时，将进入到备份交换器上绑定的队列中

TTL，即rabbitmq可以对消息或者队列设置过期时间（即在队列中的存活时间,默认消息永不过期），当两者都设置的时候，取最小值，一旦一条消息超过设置的TTL值，将会变成死信，消费者无法再收到该条消息（不是绝对的）

DLX(Dead-letter-Exchange),死信交换器，绑定DLX的队列则会死信队列，一般消息变成死信由以下几种情况发生导致：

*  消息被拒绝（Basic.reject/basic.Nack），并且设置requeue参数为false
 *  消息过期
 *  队列达到最大长度

当队列中存在死信，rabbitmq将会自动发送到设置的DLX上，进而到对应的死信队列中去

延迟队列，存储的是延迟消息，即需要等待特定的时间后，消息才能被消费者消费。实现的原理就是一个正常队列设置一个消息过期时间，以及一个对应的DLX，当过期时间达到的时候，进入DLX绑定的死信队列中，则死信队列中的消息相当于都是延迟消息

优先队列,可以通过设置队列的x-max-priority参数实现,默认优先级最低为0,最高为队列设置的最大优先级

持久化,可以提高rabbitmq的可靠性，防止在异常情况下的数据丢失（重启，宕机，关闭),持久化主要分为交换器持久化，队列持久化，消息持久化三部分

*  交换器持久化：通过声明队列时将durable参数为true进行实现，如果交换器不设置为持久化，rabbit重启后消息不会丢失，但是消息不能发送到该交换器了，对于一个长期使用的交换器建议持久化
 *  队列持久化：通过声明队列时将durable参数为true实现，如果队列不设置持久化，重启后队列的相关元数据会丢失，同时数据也会丢失,队列持久化能保证其相关元数据不会因为异常情况而丢失，但不保证消息不会丢失
 *  消息持久化: 需要将消息投递模式中BasicProperties中的deliveryMode属性设置为2即可实现

即使全部持久化，数据也有可能丢失，如消费者收到消息后还没有来得及处理就宕机了，这种情况可以设置autoAck参数为false(即要有消费者的确认后才能删除消息)；rabbitmq持久化数据到磁盘时发生故障导致数据没有及时落盘，这时候可以通过配置镜像队列来进行高可用，一般关键业务的队列数据都需要进行配置。还可以通过发送端引入事务机制或者确认机制来确保数据正常

生产者确认,生产者将消息发出之后，有可能没有正常到达服务器，即到达之前可能已经丢失了，这种情况持久化解决不了,可以通过以下方式解决：

*  事务机制:channel.txSelect(事务开启) channel.txCommit(事务提交）channel.txRollback(事务回滚) ，如果有异常情况，我们可以捕捉后通过channel.txRollback进行回滚,从而进行消息的重发,该方式的最大问题就是同步阻塞，严重影响性能
 *  发送方确认机制: 发送方将信道设置为confirm模式(channel.confirmSelect),发送方发送一条消息后在信道中等待确认的返回，同时进行下一条消息的发送，当消息最终确认后，发送方可以通过回调的方法来处理该确认消息(确认以ack或nack进行返回，nack即代表rabbitmq因为自身原因导致消息丢失)，注意尽量批量confirm,调用channel.waitForConfirms或者异步confirm提供一个回调方法，这样能够提供性能

消费端要点:

*  消息分发：当队列拥有多个消费者的时候，消息将以轮询的形式分发给消费者。那么如何防止消费者任务繁重呢？通过使用channel.basicQos(int prefetchCount,设置为0则无上限)，即允许信道上消费者所能保持的最大未确认消息，当发送一条消息给消费者时，计数加1，如果达到所设定的上限，则rabbitmq不会给该消费者发送任何消息，直到消费者确认了某条消息，计数减一(该模式对应拉模式无效)
 *  消息顺序性:很多情况下消息会错序，如事务回滚，优先级队列，延迟队列等，要保证消息顺序性，需要业务方在消息体内添加全部有序标识
 *  弃用QueueingConsumer:QueueingConsumer会拖累同一个Connection下的所有信道，使其性能降低；Rabbitmq的自动连接恢复机制不支持QueueingConsumer的这种形式;QueueingConsumer不是事件驱动的。为了避免不必要的麻烦，建议消费的时候尽量继承DefaultConsumer的方式

消息传输保障:

传输保障分为三个等级:

*  At most once:最多一次，消息可能会丢失，但绝不会重复传输
 *  At least once:最少一次，消息绝不会丢失，但可能重复
 *  Exactly once:恰好一次,每条消息肯定会被传输一次且仅一次

最少一次需要考虑:

*  消息生产者需要开启事务机制或者publisher confirm机制，以确保消息可以可靠地传出到rabbitmq中
 *  消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能从交换器路由到队列中，防止被丢失
 *  消息和队列都需要进行持久化，以确保RabbitMQ服务器在遇到异常情况不会消息丢失
 *  消费者在消费消息同时需要将autoAck设置为false，然后通过手动确认的方式去确认已经正确消费的消息，以避免不必要的消息丢失

最多一次则是生产者随意发，消费者随意消费，这很难保证消息不会丢失

恰好一次目前Rabbitmq是无法保证的。

rabbitmq集群实现原理

rabbitmq由erlang编写，Erlang天生具有分布式特性，所以rabbitmq天然支持集群。

交换器的元数据信息在所有节点上是一致的，而Queue的完整数据只会存在于它所创建的那个节点上，其他节点只知道这个queue的元数据和一个指向该queue的owner node指针

RabbitMQ集群会始终同步四种类型的内部元数据（类似索引）：

a.队列元数据：队列名称和它的属性；

b.交换器元数据：交换器名称、类型和属性；

c.绑定元数据：一张简单的表格展示了如何将消息路由到队列；

d.vhost元数据：为vhost内的队列、交换器和绑定提供命名空间和安全属性；

因此，当用户访问其中任何一个RabbitMQ节点时，通过rabbitmqctl查询到的queue／user／exchange/vhost等信息都是相同的。

HAProxy提供高可用性、负载均衡以及基于TCP和HTTP应用的代理，支持虚拟主机，它是免费、快速并且可靠的一种解决方案。根据官方数据，其最高极限支持10G的并发。HAProxy支持从4层至7层的网络交换，即覆盖所有的TCP协议。就是说，Haproxy 甚至还支持 Mysql 的均衡负载。为了实现RabbitMQ集群的软负载均衡，这里可以选择HAProxy。

通过haproxy来进行rabbitmq集群中节点的故障检查,发送故障后需要多少次健康检查才能再次确认可用，经理多少次失败健康检查后，HaProxy才会停止使用该节点服务。