rabbitMQ的学习第一篇(持续更新)

2024-03-06 11:58:01

rabbitMQ的学习目标

需要带着目的去学习一项技术,在学习中深入思考感受,
1:mq的使用场景
2:几种mq什么时候选择rabbitMQ
3:MQ保存和消费消息的性能于其他MQ的对比
4:系统的qps,系统的承受能力
5;消息的投递与消费:
1.正确投递
2消息正确消费
3. 消息幂等性
4. 消息的分布式存储
使用实例:在高并发情况下,添加商品,更新倒排索引和更新页面静态化数据需要一起执行,不仅需要严格控制事务,还要保证执行效率.添加商品的并发比较大,效率极其低下.如果采用多线程的方式田勇,也是基于一定可见时间内的调用.也会出现更新失败的情况.
这时需要考虑到异步的调用,需要容忍数据的在一定时间内的暂时不一致,商品更新成功后把执行后面任务的消息存储到公共位置这里就是消息中间键,在启用其他的服务获取消息执行后面任务.这样对于高并发情况下,商品的添加性能大大提高.
这样又引发了新的问题,保存消息的服务器的高可用,安全,性能,消息的准确投递,消息的消费等一系列问题,后续业务处理也可能会失败,失败了如何处理.

分布式异步通信模式

比较典型的模式为,生产者消费者模式,可以跨平台、支持异构系统，通常借助消息中间件来完成。

消息中间键的简介:

消息中间件也可以称消息队列，是指用高效可靠的消息传递机制进行与平台无关的数据交流，并基
于数据通信来进行分布式系统的集成。通过提供消息传递和消息队列模型，可以在分布式环境下扩展进
程的通信。
消息中间件就是在通信的上下游之间截断：break it，Broker
然后利用中间件解耦、异步的特性，构建弹性、可靠、稳定的系统。
异步处理、流量削峰、限流、缓冲、排队、最终一致性、消息驱动等需求的场景都可以使用消息中间件。

主流消息中间件及选型

当前业界比较流行的开源消息中间件包括：ActiveMQ、RabbitMQ、RocketMQ、Kafka、ZeroMQ等，其中应用最为广泛的要数RabbitMQ、RocketMQ、Kafka这三款。

选取原则

首先，产品应该是开源的。开源意味着如果队列使用中遇到bug，可以很快修改，而不用等待开发
者的更新。
其次，产品必须是近几年比较流行的，要有一个活跃的社区。这样遇到问题很快就可以找到解决方
法。同时流行也意味着bug较少。流行的产品一般跟周边系统兼容性比较好。
最后，作为消息队列，要具备以下几个特性：
1、消息传输的可靠性：保证消息不会丢失。
2、支持集群，包括横向扩展，单点故障都可以解决。
3、性能要好，要能够满足业务的性能需求。

三款MQ的对比

RabbitMQ

RabbitMQ开始是用在电信业务的可靠通信的，也是少有的几款支持AMQP协议的产品之一。
优点：

轻量级，快速，部署使用方便
支持灵活的路由配置。RabbitMQ中，在生产者和队列之间有一个交换器模块。根据配置的路
由规则，生产者发送的消息可以发送到不同的队列中。路由规则很灵活，还可以自己实现。
RabbitMQ的客户端支持大多数的编程语言。
缺点：
如果有大量消息堆积在队列中，性能会急剧下降
RabbitMQ的性能在Kafka和RocketMQ中是最差的，每秒处理几万到几十万的消息。如果应
用要求高的性能，不要选择RabbitMQ。 3. RabbitMQ是Erlang开发的，功能扩展和二次开发代价很高。

RocketMQ

RocketMQ是一个开源的消息队列，使用java实现。借鉴了Kafka的设计并做了很多改进。
RocketMQ主要用于有序，事务，流计算，消息推送，日志流处理，binlog分发等场景。经过了历次的
双11考验，性能，稳定性可可靠性没的说。
RocketMQ几乎具备了消息队列应该具备的所有特性和功能。
java开发，阅读源代码、扩展、二次开发很方便。
对电商领域的响应延迟做了很多优化。在大多数情况下，响应在毫秒级。如果应用很关注响应时
间，可以使用RocketMQ。
性能比RabbitMQ高一个数量级，每秒处理几十万的消息。
缺点：
跟周边系统的整合和兼容不是很好。

Kafka

Kafka的可靠性，稳定性和功能特性基本满足大多数的应用场景。
跟周边系统的兼容性是数一数二的，尤其是大数据和流计算领域，几乎所有相关的开源软件都支持
Kafka。
Kafka高效，可伸缩，消息持久化。支持分区、副本和容错。
Kafka是Scala和Java开发的，对批处理和异步处理做了大量的设计，因此Kafka可以得到非常高的
性能。它的异步消息的发送和接收是三个中最好的，但是跟RocketMQ拉不开数量级，每秒处理几十万
的消息。
如果是异步消息，并且开启了压缩，Kafka最终可以达到每秒处理2000w消息的级别。
但是由于是异步的和批处理的，延迟也会高，不适合电商场景。

rabbitmq的优势就是轻量型,配置简单,使用方便,延迟小,主题数量对于延迟没有影响.

削去秒杀场景下的峰值写流量

将秒杀请求暂存于消息队列，业务服务器响应用户“秒杀结果正在处理中。。。”，释放系统资源去
处理其它用户的请求。
削峰填谷，削平短暂的流量高峰，消息堆积会造成请求延迟处理，但秒杀用户对于短暂延迟有一定
容忍度。
秒杀商品有 1000 件，处理一次购买请求的时间是 500ms，那么总共就需要 500s 的时间。这时你
部署 10 个队列处理程序，那么秒杀请求的处理时间就是 50s，也就是说用户需要等待 50s 才可以看到
秒杀的结果，这是可以接受的。这时会并发 10 个请求到达数据库，并不会对数据库造成很大的压力。
通过异步处理简化秒杀请求中的业务流程
先处理主要的业务，异步处理次要的业务。
如主要流程是生成订单、扣减库存；次要流程比如购买成功之后会给用户发优惠券，增加用户的积
分。
此时秒杀只要处理生成订单，扣减库存的耗时，发放优惠券、增加用户积分异步去处理了。
解耦，实现秒杀系统模块之间松耦合
将秒杀数据同步给数据团队，有两种思路：

使用 HTTP 或者 RPC 同步调用，即提供一个接口，实时将数据推送给数据服务。
系统的耦合度高，如果其中一个服务有问题，可能会导致另一个服务不可用。
使用消息队列
将数据全部发送给消息队列，然后数据服务订阅这个消息队列，接收数据进行处理。

Java消息服务应用程序结构支持两种模式：

JMS协议即Java消息服务

jms（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间
件（MOM，Message oriented Middleware）的API，用于在两个应用程序之间，或分布式系统中发送
消息，进行异步通信。与具体平台无关的API，绝大多数MOM提供商都支持。
它类似于JDBC(Java Database Connectivity)。
包括报文和消息主题,以下为报文信息

点对点也叫队列模式
发布/订阅模式
1、在点对点或队列模型下
一个生产者向一个特定的队列发布消息，一个消费者从该队列中读取消息。这里，生产者知道消费
者的队列，并直接将消息发送到消费者的队列，概括为：
一条消息只有一个消费者获得
生产者无需在接收者消费该消息期间处于运行状态，接收者也同样无需在消息发送时处于运行
状态。
每一个成功处理的消息要么自动确认，要么由接收者手动确认。
2、发布/订阅模式
支持向一个特定的主题发布消息。
0或多个订阅者可能对接收特定消息主题的消息感兴趣。
发布者和订阅者彼此不知道对方。
多个消费者可以获得消息
在发布者和订阅者之间存在时间依赖性。
发布者需要建立一个主题，以便客户能够订阅。
订阅者必须保持持续的活动状态以接收消息，否则会丢失未上线时的消息。
对于持久订阅，订阅者未连接时发布的消息将在订阅者重连时重发。

JMS在应用集群中的问题

生产中应用基本上都是以集群部署的。在Queue模式下，消息的消费没有什么问题，因为不同节点
的相同应用会抢占式地消费消息，这样还能分摊负载。
如果使用Topic广播模式？对于一个消息，不同节点的相同应用都会收到该消息，进行相应的操
作，这样就重复消费了。。。

方案一：选择Queue模式，创建多个一样的Queue，每个应用消费自己的Queue。
弊端：浪费空间，生产者还需要关注下游到底有几个消费者，违反了“解耦”的初衷。
方案二：选择Topic模式，在业务上做散列，或者通过分布式锁等方式来实现不同节点间的竞争。
弊端：对业务侵入较大，不是优雅的解决方法。
JMS是JEE平台的标准消息传递API。它可以在商业和开源实现中使用。每个实现都包括一个JMS服
务器，一个JMS客户端库，以及用于管理消息传递系统的其他特定于实现的组件。 JMS提供程序可以是
消息传递服务的独立实现，也可以是非JMS消息传递系统的桥梁。
JMS客户端API是标准化的，因此JMS应用程序可在供应商的实现之间移植。但是：

底层消息传递实现未指定，因此JMS实现之间没有互操作性。除非存在桥接技术，否则想要共
享消息传递的Java应用程序必须全部使用相同的JMS实现。
如果没有供应商特定的JMS客户端库来启用互操作性，则非Java应用程序将无法访问JMS。 3. AMQP 0-9-1是一种消息传递协议，而不是像JMS这样的API。任何实现该协议的客户端都可以
访问支持AMQP 0-9-1的代理。
协议级的互操作性允许以任何编程语言编写且在任何操作系统上运行的AMQP 0-9-1客户端都
可以参与消息传递系统，而无需桥接不兼容的服务器实现

AMQP协议剖析

AMQP全称高级消息队列协议（Advanced Message Queuing Protocol），是一种标准，类似于
JMS，兼容JMS协议。目前RabbitMQ主流支持AMQP 0-9-1，3.8.4版本支持AMQP 1.0。

AMQP中的概念

Publisher：消息发送者，将消息发送到Exchange并指定RoutingKey，以便queue可以接收到指
定的消息。
Consumer：消息消费者，从queue获取消息，一个Consumer可以订阅多个queue以从多个
queue中接收消息。
Server：一个具体的MQ服务实例，也称为Broker。
Virtual host：虚拟主机，一个Server下可以有多个虚拟主机，用于隔离不同项目，一个Virtual
host通常包含多个Exchange、Message Queue。
Exchange：交换器，接收Producer发送来的消息，把消息转发到对应的Message Queue中。
Routing key：路由键，用于指定消息路由规则（Exchange将消息路由到具体的queue中），通
常需要和具体的Exchange类型、Binding的Routing key结合起来使用。
Bindings：指定了Exchange和Queue之间的绑定关系。Exchange根据消息的Routing key和
Binding配置（绑定关系、Binding、Routing key等）来决定把消息分派到哪些具体的queue中。这依
赖于Exchange类型。
Message Queue：实际存储消息的容器，并把消息传递给最终的Consumer。

AMQP 传输层架构

AMQP是一个二进制的协议，信息被组织成数据帧，有很多类型。数据帧携带协议方法和其他信
息。所有数据帧都拥有基本相同的格式：帧头，负载，帧尾。数据帧负载的格式依赖于数据帧的类型。
我们假定有一个可靠的面向流的网络传输层（TCP/IP或等价的协议）。
在一个单一的socket连接中，可能有多个相互独立的控制线程，称为“channel”。每个数据帧使用
通道号码编号。通过数据帧的交织，不同的通道共享一个连接。对于任意给定通道，数据帧严格按照序
列传输。
我们使用小的数据类型来构造数据帧，如bit，integer，string以及字段表。数据帧的字段做了轻微
的封装，不会让传输变慢或解析困难。根据协议规范机械地生成成数据帧层相对简单。
线级别的格式被设计为可伸缩和足够通用，以支持任意的高层协议（不仅是AMQP）。我们假定
AMQP会扩展，改进以及随时间的其他变化，并要求wire-level格式支持这些变化。
3.3.3.2 数据类型
AMQP 使用的数据类型如下：
Integers（数值范围1-8的十进制数字）：用于表示大小，数量，限制等，整数类型无符号
的，可以在帧内不对齐。
Bits（统一为8个字节）：用于表示开/关值。
Short strings：用于保存简短的文本属性，字符串个数限制为255，8个字节
Long strings：用于保存二进制数据块。
Field tables：包含键值对，字段值一般为字符串，整数等。

AMQP客户端实现JMS客户端

RabbitMQ的JMS客户端用RabbitMQ Java客户端实现，既与JMS API兼容，也与AMQP 0-9-1协议兼
容。
局限性
RabbitMQ JMS客户端不支持某些JMS 1.1功能：
JMS客户端不支持服务器会话。
XA事务支持接口未实现。
RabbitMQ JMS主题选择器插件支持主题选择器。队列选择器尚未实现。
支持RabbitMQ连接的SSL和套接字选项，但仅使用RabbitMQ客户端提供的（默认）SSL连接
协议。
RabbitMQ不支持JMS NoLocal订阅功能，该功能禁止消费者接收通过消费者自己的连接发布
的消息。可以调用包含NoLocal参数的方法，但该方法将被忽略。
RabbitMQ使用amqp协议，JMS规范仅对于Java的使用作出的规定，跟其他语言无关，协议是语言
无关的，只要语言实现了该协议，就可以做客户端。如此，则不同语言之间互操作性得以保证。

RabbitMQ介绍、概念、基本架构

RabbitMQ，俗称“兔子MQ”（可见其轻巧，敏捷），是目前非常热门的一款开源消息中间件，不管
是互联网行业还是传统行业都广泛使用（最早是为了解决电信行业系统之间的可靠通信而设计）。

高可靠性、易扩展、高可用、功能丰富等
支持大多数（甚至冷门）的编程语言客户端。
RabbitMQ遵循AMQP协议，自身采用Erlang（一种由爱立信开发的通用面向并发编程的语
言）编写。
RabbitMQ也支持MQTT等其他协议。
RabbitMQ具有很强大的插件扩展能力，官方和社区提供了非常丰富的插件可供选择：
https://www.rabbitmq.com/community-plugins.html

RabbitMQ常用的交换器类型有

请看我上一篇文章:https://blog.csdn.net/weixin_44605216/article/details/119253642

RabbitMQ数据存储

存储机制
RabbitMQ消息有两种类型：

持久化消息和非持久化消息。
这两种消息都会被写入磁盘。
持久化消息在到达队列时写入磁盘，同时会内存中保存一份备份，当内存吃紧时，消息从内存中清
除。这会提高一定的性能。
非持久化消息一般只存于内存中，当内存压力大时数据刷盘处理，以节省内存空间。
RabbitMQ存储层包含两个部分：队列索引和消息存储。

队列索引：rabbit_queue_index

索引维护队列的落盘消息的信息，如存储地点、是否已被给消费者接收、是否已被消费者ack等。
每个队列都有相对应的索引。
索引使用顺序的段文件来存储，后缀为.idx，文件名从0开始累加，每个段文件中包含固定的
segment_entry_count 条记录，默认值是16384。每个index从磁盘中读取消息的时候，至少要在内存
中维护一个段文件，所以设置 queue_index_embed_msgs_below 值得时候要格外谨慎，一点点增大也
可能会引起内存爆炸式增长。

消息存储：rabbit_msg_store

消息以键值对的形式存储到文件中，一个虚拟主机上的所有队列使用同一块存储，每个节点只有一
个。存储分为持久化存储（msg_store_persistent）和短暂存储（msg_store_transient）。持久化存
储的内容在broker重启后不会丢失，短暂存储的内容在broker重启后丢失。
store使用文件来存储，后缀为.rdq，经过store处理的所有消息都会以追加的方式写入到该文件
中，当该文件的大小超过指定的限制（file_size_limit）后，将会关闭该文件并创建一个新的文件以供新
的消息写入。文件名从0开始进行累加。在进行消息的存储时，RabbitMQ会在ETS（Erlang Term
Storage）表中记录消息在文件中的位置映射和文件的相关信息。
消息（包括消息头、消息体、属性）可以直接存储在index中，也可以存储在store中。最佳的方式
是较小的消息存在index中，而较大的消息存在store中。这个消息大小的界定可以通过
queue_index_embed_msgs_below 来配置，默认值为4096B。当一个消息小于设定的大小阈值时，就
可以存储在index中，这样性能上可以得到优化。一个完整的消息大小小于这个值，就放到索引中，否
则放到持久化消息文件中。

读取消息时，先根据消息的ID（msg_id）找到对应存储的文件，如果文件存在并且未被锁住，则直
接打开文件，从指定位置读取消息内容。如果文件不存在或者被锁住了，则发送请求由store进行处
理。
删除消息时，只是从ETS表删除指定消息的相关信息，同时更新消息对应的存储文件和相关信息。
在执行消息删除操作时，并不立即对文件中的消息进行删除，也就是说消息依然在文件中，仅仅是标记
为垃圾数据而已。当一个文件中都是垃圾数据时可以将这个文件删除。当检测到前后两个文件中的有效
数据可以合并成一个文件，并且所有的垃圾数据的大小和所有文件（至少有3个文件存在的情况下）的
数据大小的比值超过设置的阈值garbage_fraction（默认值0.5）时，才会触发垃圾回收，将这两个文件
合并，执行合并的两个文件一定是逻辑上相邻的两个文件。合并逻辑：

锁定这两个文件
先整理前面的文件的有效数据，再整理后面的文件的有效数据
将后面文件的有效数据写入到前面的文件中
更新消息在ETS表中的记录
删除后面文件

如果消息投递的目的队列是空的，并且有消费者订阅了这个队列，那么该消息会直接发送给消费
者，不会经过队列这一步。当消息无法直接投递给消费者时，需要暂时将消息存入队列，以便重新投
递。

队列结构

rabbit_variable_queue.erl 源码中定义了RabbitMQ队列的4种状态：

alpha：消息索引和消息内容都存内存，最耗内存，很少消耗CPU
beta：消息索引存内存，消息内存存磁盘
gama：消息索引内存和磁盘都有，消息内容存磁盘
delta：消息索引和内容都存磁盘，基本不消耗内存，消耗更多CPU和I/O操作

 	消息存入队列后，不是固定不变的，它会随着系统的负载在队列中不断流动，消息的状态会不断发

送变化。
持久化的消息，索引和内容都必须先保存在磁盘上，才会处于上述状态中的一种
gama状态只有持久化消息才会有的状态。
在运行时，RabbitMQ会根据消息传递的速度定期计算一个当前内存中能够保存的最大消息数量
（target_ram_count），如果alpha状态的消息数量大于此值，则会引起消息的状态转换，多余的消息
可能会转换到beta、gama或者delta状态。区分这4种状态的主要作用是满足不同的内存和CPU需求。
通常队列由rabbit_amqqueue_process和backing_queue这两部分组成，
rabbit_amqqueue_process负责协议相关的消息处理，即接收生产者发布的消息、向消费者交付消
息、处理消息的确认（包括生产端的confirm和消费端的ack）等。backing_queue是消息存储的具体形
式和引擎，并向rabbit_amqqueue_process提供相关的接口以供调用。

对于普通没有设置优先级和镜像的队列来说，backing_queue的默认实现是rabbit_variable_queue，
其内部通过5个子队列Q1、Q2、delta、Q3、Q4来体现消息的各个状态

阿尔法(a):索引和消息都在内存里
βγ:索引在内存中,消息在磁盘中
δ:索引和消息都在磁盘中
消息的存储会根据服务器的cpu和内存状态会改变消息的状态.
消费者获取消息也会引起消息的状态转换.

首先会从Q4中获取消息，如果获取成功则返回。
如果Q4为空，则尝试从Q3中获取消息，系统首先会判断Q3是否为空，如果为空则返回队列
为空，即此时队列中无消息。
如果Q3不为空，则取出Q3中的消息；进而再判断此时Q3和Delta中的长度，如果都为空，则
可以认为 Q2、Delta、 Q3、Q4 全部为空，此时将Q1中的消息直接转移至Q4，下次直接从
Q4 中获取消息。
如果Q3为空，Delta不为空，则将Delta的消息转移至Q3中，下次可以直接从Q3中获取消息。
在将消息从Delta转移到Q3的过程中，是按照索引分段读取的，首先读取某一段，然后判断读
取的消息的个数与Delta中消息的个数是否相等，如果相等，则可以判定此时Delta中己无消
息，则直接将Q2和刚读取到的消息一并放入到Q3中，如果不相等，仅将此次读取到的消息转
移到Q3。