rabbit
2024-05-06 16:54:21 49 举报AI智能生成
rabbitMQ 基础
作者其他创作
大纲/内容
解耦
冗余
扩展性
削峰
可恢复性
顺序性
缓冲
异步通信
作用
优点：延迟性低 后台管理友好 社区比较活跃
缺点：吞吐量低erlang语言开发，深入源码不太方便
优劣
可靠性：RabbitMQ 使用 些机制来保证可靠性 如持久化、传输确认及发布确认等
灵活的路由：在消息进入队列之前，通过交换器来路由消息 对于典型的路由功能，RabbitMQ 己经提供了 些内置的交换器来实现 针对更复杂的路由功能，可以将多个交换器绑定在 起， 可以通过插件机制来实现自己的交换器
扩展性：多个 MQ 节点可以组成 个集群，也可以根据实际业务情况动态地扩展集群中节点。
高可用：队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用
Module Laye 位于协议最高层，主要定义了一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑。例如 客户端可以使用 Queue Declare 命令声明个队列或者使用 Basic Consume订阅消费一个队列中的消息。Session Layer：位于中间层，主要负责将客户端的命令发送给服务器 再将服务端的应答返回给客户端，主要为客户端与服务器之间的通信提供可靠性同步机制和错误处理。Tranport Laye 位于最底层，主要传输二进制数据流，提供帧的处理、信道复用、错误检测和数据表示等。
AMQP
多种协议（默认AMQP）
多语言客户端
管理界面
插件机制
具体特点
生产者连接到 RabbitMQ Broker 建立一个连接（ onnection ），开启一个信道（参考3.1）生产者声明一个交换器，并设置相关属性，比如交换机类型、是否持久化等参考3.2）生产者声明 个队列井设置相关属性，比如是否排他、是否持久化、是否自动删除等（参考3.2）生产者通过路由键将交换器和队列绑定起来（参考3.2）生产者发送消息至 RabbitMQ Broker，其中包含路由键、交换器等信息(参考3.3)相应的交换器根据接收到的路由键查找相匹配的队列如果找到，则将从生产者发送过来的消息存入相应的队列中。如果没有找到 ，则根据生产者配置的属性选择丢弃还是回退给生产者（参考4.1）关闭信道。关闭连接。
Producer：生产者
消费者连接到 RabbitMQ Broker ，建立一个连接（ Connection ），开启一个信道（ Channel)消费者向 RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数，以及做 些准备工作（详细内容请参考 3.4 节〉。等待 RabbitMQ Broker 回应并投递相应队列中的消息，消费者接收消息。消费者确认（ ack ）接收到的消息RabbitMQ 从队列中删除相应己经被确认的消息关闭信道。
Consumer：消费者
Broker：服务节点（一个Broker可以看做是一个服务节点或者服务实例）
架构
Connection：连接
连接tcp连接复用，不仅可以减少性能开销，同时也便于管理。每个线程都把持一个信道，所以信道复用了TCP连接。同时rabbitmq可以确保每个线程的私密性，就像拥有独立的连接一样。当每个信道的流量不是很大时，复用单一的connection可以再产生性能瓶颈的情况下有效地节省tcp连接资源，但是当信道本身的流量很大时，这时候多个信道复用一个connection就会产生性能瓶颈，进而是整体的流量被限制了。此时就需要开辟多个connection，将这些信道均摊到这些connection中
Channel：信道
不支持队列层面广播消费
Queue：存储消息
fanout：把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中
direct：把消息路由到那些 BindingKey = RoutingKey完全匹配的队列中
topic：通过*和#进行模糊匹配，没有direct那么绝对相等
headers：不依赖路由键的匹配规则，依赖依赖消息内容中的headers属性。性能差，很少使用
交换器类型
Exchange：交换器
需要与交换器类型和绑定键（BindingKey）联合使用生效
RoutingKey：路由键
在发送消息的时候，其中需要的路由键是 RoutingKey
在使用绑定的时候，其中需要的路由键是 BindingKey
大多数情况下习惯性地将 BindingKey写成 RoutingKey，如何区分
当BindingKey和RoutingKey相匹配时，消息会路由到对应队列中
Binding：绑定键（BindingKey）
开箱即用的默认vhost:\"/\"
权限隔离
vhost（虚拟主机）
组件介绍
监听Basic.Return
addReturnListener：添加ReturnListener监听器
exchange：交换机名称
type：交换机类型
durable：设置是否持久化
自动删除的前提是至少有 个队列或者交换器与这个交换器绑定 之后所有与这个交换器绑定的队列或者交换器都与 解绑。注意不能错误地 这个参数理解为 “当与此交换器连接的客户端都断开时 RabbitMQ 会自动 除本交换器
autoDelete：设置是否自动删除
如果设置为 true ，则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换机路由到交换机这种方式。
internal：设置是否内置
alternate-exchange：绑定备份交换机
argument：其他一些结构化参数
exchangeDeclare：声明交换机
exchangeDeclareNoWait方法：不常用，不建议，比exchangeDeclare多一个NoWait属性
exchangeDelete：删除交换机
无参声明队列：由 RabbitMQ 命名的（类似这种amq.gen-LhQzlgv3GhDOv8PIDabOXA 名称，这种队列也称之为匿名队列〉、排他的、自动删除的、非持久化的队列。
queue：队列名称
如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点：排他队列是基于连接（ Connection ）可见的，同 个连接的不同信道（ Channel)是可以同时访问同一连接创建的排他队列：“首次”是指如果 个连接己经声明了排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同：即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景。
exclusive：设置是否排他
为 true 则设置队列为自动删除。自动删除的前提是：至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。不能把这个参数错误地理解为：“当连接到此队列的所有客户端断开时，这个队列自动删除”，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列。
有参声明队列
注意：生产者和消费者都能够使用 queueDeclare 来声明一个队列，但是如果消费者在同一个信道上订阅了另一个队列，就无法再声明队列了。必须先取消订阅，然后将信道直为“传输”模式，之后才能声明队列。
queueDeclare：声明队列
这个方法用来检测相应的队列是否存在。 如果存在 正常返回 ，如果不存在则抛出异常： 404 channel exception ，同时Channel 也会被关闭
queueDeclarePassive 的方法，比较常用
queueDeclareNoWait方法：不常用，不建议，比queueDeclare多了一个NoWiat属性
queueDelete：删除队列
queuePurge：清空队列中内容，不删除队列
routingKey：用来绑定队列和交换机的路由键
queueBind：队列和交换机绑定
queueUnBind：解绑队列和交换机
exchangeBind：交换机和交换机绑定
exchange：交换机
routingKey：路由键
其包含 14 个属性成员
contentType
contentEnconding
deliveryMode：投递模式
priority：优先级
correlationid：用来关联请求（ request ）和其调用 RPC 之后的回复（ response）
replyTo：设置回调队列
expiration：过期时间
messageId
timestamp
type
userId
appId
cluster
props：消息的基本属性集
byte[] body: 消息体
true(交换器无法根据自身的类型和路由键找到一个符合条件的队列，那么 RabbitM 会调用 Basic.Return 命令将消息返回给生产者)false(上述情况，直接丢弃)
mandatory
RabbitMQ 3.0开始去掉这个参数的支持，建议采用TTL、DLX
immediate
basicPublish：发送消息
autoAck：自动确认true/手动确认false
consumerTag ：消费者标签，用来区分多个消费者
noLocal：设置为True，同一个connection的生产者发送的消息，消费者不能接收
exclusive ：设置是否排他
arguments 设置消费者的其他参数
callback 设置消费者的回调函数
basicConsumer
推（Push）模式
basicGet
拉（Pull）模式
channel.basicQos(4);当consumer接收到4条消息后，不再会接收其它消息，只有当这些消息被消费时，才会接收新的消息。
Basic Consume 将信道（ Channel ）直为接收模式，直到取消队列的订阅为止。在接收模式期间， RabbitMQ 会不断地推送消息给消费者，当然推送消息的个数还是会受到 Basic.Qos的限制．如果只想从队列获得单条消息而不是持续订阅，建议还是使用 Basic.Get 进行消费．但是不能将 Basic.Get 放在一个循环里来代替 Basic.Consume ，这样做会严重影响 RabbitMQ的性能．如果要实现高吞吐量，消费者理应使用 Basic.Consume 方法。
推拉模式的场景
deliveryTag：消息的编号
requeue：true（重新存入队列）/false（队列中移除）
basicReject（单个）
multiple：true（拒绝消息编号前的所有未被当前消费者确认的消息）
basicNack（批量）
拒绝
请求 RabbitMQ 重新发送还未被确认的消息
requeue 参数设置为 true （默认），则未被确认的消息会被重新加入到队列中，这样对于同一条消息来说，可能会被分配给与之前不同的消费者。
basicRecover
重推
消费模式
getCloseReason 方法可以让你知道对象关闭的原因
isOpen 方法检测对象当前是否处于开启状态
isHardError 方法可以知道是 Co 且口 ectio 丑的还是 Channel 的错误
getReason 方法可以获取 cause 相关的信息
ShutdownSignalException
ShutdownListener
关闭（connection和channel）
组件基础
生产者在发送消息的时候如果不设置 mandatory 参数，那么消息在未被路由的情况下将会丢失：如果设置了 mandatory 参数，那么需要添加 ReturnListener 的编程逻辑，生产者的代码将变得复杂
解决什么？
建议设置为 fanout 类型，否则可能会导致消息丢失
exchangeDeclare中参数argument设置alternate-exchange 绑定交换机
备份交换机出现（不存在、未绑定队列、未匹配队列）这些情况，不会有任何异常
如果备份交换器和 mandatory 参数一起使用，那么 mandatory 参数无效
备份交换机
RabbitMQ 不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费连接是否己经断开，这么设计的原因是 RabbitMQ允许消费者消费一条消息的时间可以很久很久。
默认无过期
TTL 设置为 表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃
queueDeclare 方法中加入x-message -ttl 参数实现的，单位毫秒
通过队列属性设置，队列中所有消息都有相同的过期时间
因为每条消息的过期时间不同，消息过期，不会马上从队列中抹去，删除所有过期消息势必要扫描整个队列，所以每条消息是否过期是在即将投递到消费者之前判定的
basicPublish 方法中加入 expiration的属性参数，单位毫秒
对消息本身进行单独设置，每条消息的 TTL 可以不同
则消息的 TTL 以两者之间较小的那个数值为准
消息在队列中的生存时间一旦超过设置 TTL 值时，就会变成“死信”
注意
设置消息的TTL
未使用的意思是队列上没有任何的消费者，队列也没有被重新声明，并且在过期时间段内也未调用过 Basic Get 命令。
通过 channel queueDeclare 方法中的 x-expires 参数可以控制队列被自动删除前处于未使用状态的时间
RabbitMQ 会确保在过期时间到达后将队列删除，但是不保障删除的动作有多及时 。在RabbitMQ重启后，持久化的队列的过期时间会被重新计算。
设置队列的TTL
TTL：过期时间
当消息在一个队列中变成死信（ dead message ）之后，它能被重新被发送到另一个交换器中，这个交换器就是 DLX ，绑定 DLX 的队列就称之为死信队列。
消息被拒绝（ Basic.Reject/Basic.Nack ），井且设置 requeue 参数为 false;
消息过期
队列达到最大长度
变成死信的情况
queueDeclare 方法中设置 x-dead-letter-exchange 参数来为这个队列添加 DLX
DLX:死信队列
假设一个应用中需要将每条消息都设置为 10 秒的延迟，生产者通过 exchange.normal 这个交换器将发送的消息存储在 queue.normal 这个队列中。消费者订阅的并非是 queue.normal 这个队列，而是 queue.dlx 这个队列 。当消息从 queue.normal 这个队列中过期之后被存入 queue.dlx 这个队列中，消费者就恰巧消费到了延迟 10 秒的这条消息
通过DLX 和TTL 实现延迟队列
延迟队列
RabbitMq 通过交换机发送到指定的队列，队列保证了顺序，但是多个消费者的话，需要消费者自己去保证，比如先通过分布式锁锁定，然后存储到表中或者缓存中。
channel.basicQos(1) 设置每次只处理一条消息通过手动ACK手动确认
顺序消息
设置队列的 x-max-priority 参数来实现
优先级高的消息可以被优先消费，这个也是有前提的
如果在消费者的消费速度大于生产者的速度Broker 中没有消息堆积的情况下，对发送的消息设置优先级也就没有什么实际意义
优先级队列
1. 客户端启动，创建一个匿名回调队列（由RabbitMq自动创建）
2. 客户端为RPC请求设置两个属性：replyTo用来告知RPC服务端回复请求时的目的队列，即回调队列；correlationId用来标记一个请求
3. 请求被发送到rpc_queque队列中
4. RPC服务端监听rpc_queue 队列中的请求，当请求到来时，服务当会处理并把带有结果的消息发送到客户端。接收的对列就是replyTo设定的回调队列
RPC实现
交换器的持久化
队列的持久化
消息的持久化
持久化
RabbitMQ消息追踪之Firehose
组件进阶
具体组件
RabbitMQ实战指南
0 条评论
回复 删除
下一页
