佛山网站到首页排名,成为网站开发工程师,百度广告联盟下载,广西自治区集约化网站建设要求文章目录1. 什么是消息队列1.1 消息队列概述1.2 使用消息队列的优势1.3 使用消息队列的劣势1.4 常见的消息队列产品对比2. RabbitMQ 基本概念2.1 RabbitMQ 概述2.2 RabbitMQ 的概念模型2.2.1 Message2.2.2 Publisher2.2.3 Exchange2.2.4 Binding2.2.5 Queue2.2.6 Connection2.2…
文章目录1. 什么是消息队列1.1 消息队列概述1.2 使用消息队列的优势1.3 使用消息队列的劣势1.4 常见的消息队列产品对比2. RabbitMQ 基本概念2.1 RabbitMQ 概述2.2 RabbitMQ 的概念模型2.2.1 Message2.2.2 Publisher2.2.3 Exchange2.2.4 Binding2.2.5 Queue2.2.6 Connection2.2.7 Channel2.2.8 Virtual Host2.2.9 Consumer2.2.10 Broker2.3 RabbitMQ 的六种工作模式2.3.1 工作队列模式2.3.2 发布/订阅模式2.3.3 路由模式2.3.4 主题模式2.3.5 RPC 模式2.3.6 消息头模式3. RabbitMQ 高级特性3.1 消息投递时序图3.2 消息的可靠投递3.2.1 publishConfirm 机制3.2.2 returnCallback 机制3.2.3 消息确认 ACK 机制3.2.3.1 消息确认 ACK 机制的基本概念3.2.3.2 Spring AMQP 实现手动确认 ACK 机制3.3 死信队列3.4 延时队列3.4.1 TTL 死信队列实现延时队列3.4.2 x-delay-message-exchange 插件实现延时队列4. RabbitMQ 如何解决问题4.1 使用 RabbitMQ 集群实现高可用4.2 如何保证消息不被重复消费4.3 如何确保消息不丢失4.3.1 消息从生产者到 broker 之间的防丢失4.3.2 消息在 broker 中的防丢失4.3.3 消息从 broker 到消费者之间的防丢失4.4 如何保证消息传递的顺序性1. 什么是消息队列
1.1 消息队列概述
消息队列Message Queue以下简称 MQ是在消息的传输过程中保存消息的容器多用于分布式系统服务间的通信。
1.2 使用消息队列的优势
使用消息队列主要有解耦、异步、削峰的优势。
应用解耦可以提高系统的容错性和可维护性异步提速提升用户体验和系统吞吐量削峰填谷提高系统的稳定性
1.3 使用消息队列的劣势
使用消息队列主要有系统可用性降低、复杂度提高的劣势。
系统可用性降低系统引入的外部依赖越多则系统的稳定性越低。一旦 MQ 宕机则系统中依赖 MQ 实现的功能全部不可用。所以 MQ 一定要 高可用。系统的复杂度提高引入 MQ 后需要考虑的问题会比较多且比较棘手主要有以下几个方面的问题 如何保证消息不被重复消费如何处理消息丢失的情况如何保证消息传递的顺序性
1.4 常见的消息队列产品对比
市面上常见的消息队列产品主要有 RabbitMQ 、ActiveMQ、RocketMQ、Kafka。
特性RabbitMQActiveMQRocketMQKafka单机吞吐量万级高于 ActiveMQ万级次于 RabbitMQ10 万级支持高吞吐10 万级支持高吞吐topic 数量对吞吐量的影响topic 可以达到几百/几千的级别吞吐量会有较小幅度的下降这是 RocketMQ 的一大优势在同等级旗下可以支撑大量的 topictopic 到几十/几百的级别时吞吐量会大幅度下降。在同等机器下如果要支持大规模的 topic需要增加更多的机器资源协议支持AMQP、XMPP、SMTP、STOMPAMQP、OpenWire、STOMP、REST、XMPP自定义自定义协议社区封装了 HTTP 协议支持时效性微秒us 级RabbitMQ 的最大特点延迟最低毫秒ms级毫秒ms级毫秒ms级内可用性高基于主从架构可实现高可用高基于主从架构可实现高可用非常高分布式架构非常高分布式架构一个数据多个副本少数机器宕机不会丢失数据不会导致不可用消息可靠性基本不丢有较低的概率丢失数据经过参数优化配置可以做到 0 丢失经过参数优化配置可以做到 0 丢失功能支持基于 erlang 开发并发能力很强性能极好延时最低社区活跃管理界面功能丰富老牌产品成熟度高文档多MQ 功能较为完善基于分布式架构扩展性好功能较为简单只支持简单的 MQ 功能在大数据领域的实时计算以及日志采集被大规模使用
2. RabbitMQ 基本概念
2.1 RabbitMQ 概述
RabbitMQ 是实现了高级消息队列协议AMQP的开源消息中间件。主要的特点包括
可靠性ReliabilityRabbitMQ使用一些机制来保证可靠性如持久化、传输确认、发布确认。灵活的路由Flexible Routing在消息进入队列之前通过 Exchange 来路由消息。对于典型的路由功能RabbitMQ已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能可以将多个 Exchange 绑定在一起也可以通过插件机制实现自己的 Exchange。消息集群Clustering多个 RabbitMQ服务器可以组成一个集群形成一个逻辑 Broker。高可用Highly Available Queues队列可以在集群中的机器上进行同步使得在部分节点出问题的情况下队列仍然可用。多种协议Multi-protocolRabbitMQ 支持多种消息队列协议比如 STOMP、MQTT。多语言客户端Many ClientsRabbitMQ 几乎支持所有的常用编程语言。管理界面Management UIRabbitMQ 提供了一个易用的用户界面。跟踪机制Tracing如果消息异常RabbitMQ 提供了消息跟踪机制可以方便定位问题。插件机制Plugin SystemRabbitMQ 提供插件机制可以根据自己的需求获得或自己编写插件以满足定制化需求。
2.2 RabbitMQ 的概念模型
2.2.1 Message
Message 即消息。消息是不具名的由**消息头header和消息体payload**组成。
消息体由二进制数组承载保证了传输的高效性和一定的安全性。消息头由一系列的可选属性组成。包括 路由键routing-key、消息优先级priority、**是否需要持久化delivery-mode**等组成。
2.2.2 Publisher
Publisher 即消息的生产者。也是一个向 broker 发布消息的客户端应用程序。
2.2.3 Exchange
Exchange 即交换机用来接收生产者发送的消息并将这些消息路由给与其绑定的队列。 Exchange 分发消息根据类型不同分发策略随之不同一共由四种类型
direct只有消息中的路由键Routing Key和绑定关系Binding完全匹配时交换机才会将消息分发到对应的队列中。fanout每条发送到 fanout 类型的交换机上的消息都会被分发到所有绑定到这个交换机的队列上去。fanout 类型的交换机不处理路由键的匹配逻辑只是简单地将消息分发到所有绑定的队列上去。由于在分发过程中少处理了路由键匹配的逻辑fanout 类型的交换机转发消息效率是最高的。topic在分发过程中需要处理路由键匹配的逻辑匹配逻辑如下: 单词与单词之间用 . 隔开* 匹配一个单词# 匹配 0 个或多个单词*、# 只能单独用作占位符不能与单词贴着使用 headers在分发过程中不需要处理路由键匹配的逻辑而使用**绑定参数Arguments**来进行匹配匹配逻辑如下: Routing Key 不参与路由逻辑如果绑定关系中 Arguments 中的 x-match all 时表示发送的消息属性中的 header 中的所有键值对与绑定关系中声明的 Arguments 中的所有键值对都相等才能匹配可以理解为消息属性中的 header 中的所有键值对包含绑定关系中声明的 Arguments 中的所有键值对。如果绑定关系中 Arguments 中的 x-match any 时表示发送的消息属性中的 header 中任一键值对与 Arguments 中相等就能匹配。
2.2.4 Binding
Binding 即队列与交换机的绑定关系可以理解为交换机和队列的关系中间表。Binding 包括以下两部分:
Routing Key即路由键可以理解为消息的路由关键字。在交换机类型为 direct 或 topic 时将参与路由逻辑。Arguments即绑定参数可以理解为消息的路由参数键值对的形式。在交换机类型为 Headers 时将参与路由逻辑。
2.2.5 Queue
Queue 即队列用来保存消息直到发送给消费者。它是存放消息的容器一个消息可以被投入到一个或多个队列。消息将一直呆在队列中等待消费者将其从队列中取走。
its essentially a large message buffer. —— 本质上队列就是一个巨大的消息缓冲区。RabbitMQ 官网上如是介绍 Queue。 队列有几个非常重要的属性
TTL即 Time-To-Live 属性队列的最大存活时间。通过在声明队列时设置 x-message-ttl 参数单位ms来完成。队列中的所有消息都会遵守这个时间的限制。最大长度限制顾名思义即队列的大小。这个最大长度可以是针对队列中就绪状态消息数量的限制也可以是针对队列中所有就绪状态消息的总字节数的限制且两个限制可以同时存在。注意这里说的大小都是针对队列中就绪状态的消息来进行统计的。 消息数量限制通过设置 x-max-length 来实现消息总字节数限制通过设置 x-max-length-bytes 来实现单位字节只计算消息头的字节数不计算消息头、消息属性占用的字节数 队列溢出行为当队列中的就绪状态消息超出了消息的最大长度限制后将会有队列溢出行为发生由 x-overflow 进行设置 drop-head删除队列头部的消息reject-publish丢掉最近接收的消息reject-publish-dlx将消息发往死信队列
2.2.6 Connection
Connection 即网络连接。
2.2.7 Channel
Channel 即信道多路复用连接中的一条独立的双向数据流通道。信道是建立在真是的 TCP 连接内的虚拟连接AMQP 命令都是通过信道发出去的不管是发布消息订阅队列还是接收消息这些动作都是通过信道完成的。因为对于 OS 来说建立和销毁 TCP 都是非常昂贵的开销所以引入了信道的概念以复用 TCP 连接。
2.2.8 Virtual Host
Virtual Host 即虚拟主机可以理解为虚拟消息服务器。每个 Virtual Host 相当于一个独立的 RabbitMQ 服务器每个 Virtual Host 之间的数据是相互隔离的不能互通。 可以将 Virtual Host 类比为 MySQL 中的 DataBase 的概念。
2.2.9 Consumer
Consumer 即消息的消费者表示一个从 broker 中获取消息的客户端应用程序。
2.2.10 Broker
Broker 即消息队列服务实体。
2.3 RabbitMQ 的六种工作模式
2.3.1 工作队列模式 工作队列模式工作队列模式实现
生产者和消费者需要指定同一个交换机生产者发送时指定路由名称为队列名称消费者将队列绑定至交换机上无需指定路由键 工作队列模式下多个消费者监听同一个队列的时候一条消息只会有一个消费者接收到。2.3.2 发布/订阅模式 发布订阅模式发布订阅模式实现
生产者和所有消费者需要指定同一个交换机指定交换机类型为 fanout生产者发送时无需指定 Routing Key消费者绑定各自队列到交换机上 发布订阅模式下所有绑定了交换机的队列都会收到消息但如果多个消费者监听了同一个队列一条消息还是只会有一个消费者收到。2.3.3 路由模式 路由模式路由模式实现
生产者和所有消费者需要指定同一个交换机指定交换机类型为 direct生产者发送时指定 Routing Key消费者将各自需要监听的队列通过自定义 Routing Key 绑定到交换机上 路由模式下只有与生产者指定了相同的 Routing Key 的队列才会收到消息如果两个队列绑定的 Routing Key 相同则都会收到消息。2.3.4 主题模式 主题模式主题模式实现
生产者和所有消费者需要指定同一个交换机指定交换机类型为 topic生产者发送时指定 Routing Key消费者将各自需要监听的队列通过自定义 Routing Key 匹配格式绑定到交换机上。 多个单词用 . 隔开否则视为一个单词# 匹配零个或多个单词* 只能匹配一个单词 主题模式下只有生产者的 Routing Key 与队列绑定的 Routing Key 格式匹配成功的队列才能收到消息。如果几个队列的 Routing Key 格式都能匹配上则都会收到消息。2.3.5 RPC 模式 RPC 模式RPC 模式 Spring AMQP 实现
客户端和服务端需要指定同一个交换机指定交换机类型为 direct客户端发送请求时指定 Routing Key 且使用 RabbitMQ.convertAndSendAndRecive 方法将请求入参传入此方法将返回一个 Object 对象即为 RPC 的调用结果在未获取结果之前发送的线程将一直处于阻塞状态。服务端将需要监听的队列通过指定 Routing Key 绑定到交换机上且将 RabbitListener 标注的方法的返回参数设置为 RPC 调用的返回结果。 RPC 模式可以实现 RPC 调用的功能。但在未使用 Spring AMQP 的情况下需要手动指定 RPC 的回调队列且需要手动监听回调队列并从中获取指定 correlationId 的返回结果Spring AMPQ 已经自动实现了这些功能。2.3.6 消息头模式
消息头模式与路由模式类似不过消息头模式并不使用路由键来进行匹配而是使用绑定参数来进行匹配。匹配规则见Exchange 类型之 header
3. RabbitMQ 高级特性
3.1 消息投递时序图
RabbitMQ 的消息投递时序图如下所示
3.2 消息的可靠投递
在使用消息队列时由于消息链路上涉及到三个终端则如何避免消息丢失或者投递失败是必须要考虑的问题。RabbitMQ 提供了三种机制来控制消息的投递可靠性
生产者发送确认机制发送确认分为两部分 确认生产者发送的消息是否到达 broker或者说是否到达交换机默认只要到达了 broker 就一定会到达交换机即 publishConfirm 机制确认交换机的消息是否由路由规则到达了队列即 returnCallback 机制 消费者接收确认机制确认消费者是否成功消费了队列中的消息即消息确认 ACK 机制
3.2.1 publishConfirm 机制
当消息到达 broker 之后会触发 confirmCallback 回调确认消息已经到达 broker默认只要到达了 broker 就一定会到达交换机。 在 SpringBoot 应用中需要开启配置
spring:rabbitmq:# 高版本写法publisherConfirmType: CORRELATED# 低版本写法publisher-confirms: true然后在 RabbitTemplate 中调用 setConfirmCallback 设置具体回调逻辑。
3.2.2 returnCallback 机制
当交换机按照路由规则投递消息至队列失败时会触发 returnCallback 回调将消息退回或者直接丢弃。 在 SpringBoot 应用中需要开启配置
spring:rabbitmq:publisher-returns: true 然后在 RabbitTemplate 中调用 setReturnsCallback 设置具体回调逻辑。 注意在 AMPQ 协议中mandatory 标识位用于控制消息的退回逻辑在 SpringBoot 应用中对应
当 RabbitTemplate 中的 mandatory 属性为 true 时将会将消息退回给生产者当 RabbitTemplate 中的 mandatory 属性为 false时将直接把消息丢弃
3.2.3 消息确认 ACK 机制
3.2.3.1 消息确认 ACK 机制的基本概念
消息确认 ACK 机制即消费者接收到 broker 投递的消息后与 broker 的一个应答机制。消费者接收到 broker 投递的消息后会出现以下几种情况
消费者在接收到消息后正常返回给 broker 一个 ACK 的应答broker 在接收到这个应答后将消息从队列中删除消费者自身出现了宕机/逻辑出错导致没有正常返回 broker 一个正常的应答或者使用了 nack 或者 reject 应答则根据 requeue 参数的不同执行不同的逻辑reject 应答对应的 requeue 参数值为 true 如果 requeue 参数值为 false则 broker 会直接丢弃这条消息如果 requeue 参数值为 true则会将这条消息发送给正在监听这个队列的其他消费者如果只有当前消费者则会继续把这条消息投递给当前消费者。 nack 和 reject 的主要区别在于reject 只能拒绝单个消息但 nack 可以拒绝小于指定 deliveryTag 的所有消息。deliveryTag 是 AMQP 中针对一个消息的序号值同一消息每次从 broker 发出后该序号值都会 1。注意如果在消费者的消费逻辑中没有捕获异常或者在手动确认模式下一直没有回复 ACK 那么消息将会不停地重新分发将极大地影响 broker 的运行稳定性。
3.2.3.2 Spring AMQP 实现手动确认 ACK 机制
Spring AMQP 中ACK 一共有三种确认模式
AcknowledgeMode.AUTO自动确认AcknowledgeMode.MANUAL手动确认AcknowledgeMode.NONE不确认 一般来说我们在实现消息可靠性投递时最好将确认模式修改为手动确认模式手动处理 ACK 。配置文件如下
spring:rabbitmq:listener:# spring amqp 的监听容器有两种simple 和 direct可以根据不同的场景选择不同的监听容器simple:acknowledge-mode: manual在完成上面的设置之后需要在监听方法入参中填入 Channel 对象
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;RabbitListener(//在这里完善你的绑定关系)/*** p消息监听方法/p** param header 消息头键值对* param message 消息体* param channel 信道*/public void onListening(Headers MapString, Object header, Payload Message message, Channel channel) {// do something here//消息处理完成后调用 ack 方法//这里的 deliveryTag 从消息中获得channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);// 或者可以使用 nack 方法第二个参数是是否批量处理第三个参数即为 requeue 参数//channel.basicNack(message.getMessageProperties().getDeliveryTag(), true, true);}3.3 死信队列
首先需要理解一下什么叫死信。正常的消息可以按照消息投递时序图走完整个流程但是出现以下几种情况后正常消息就变成了死信
消息被拒绝basicReject 或 basicNack且 requeue 值为 false超过了队列指定的 x-message-ttl 属性值或者超过了消息设定的 expiration 值时队列超过了最大长度限制且队列溢出行为参数为 reject-publish-dlx 时 死信队列实际上就是一个接收死信的队列其与正常的交换机队列完全一致。使用方法
在队列属性上添加 x-dead-letter-exchange 和 x-dead-letter-routing-key 参数指定死信出现时将死信转发过去的目标死信队列在队列属性上添加 x-overflow 参数并指定值为 reject-publish-dlx 死信队列应用场景
当发生消费异常时可以方便地将出错的消息发送至死信队列随后调用业务逻辑将当时出错的数据持久化到数据库或者执行其他的操作。通过这样的方式不用去翻看应用日志就可以知道找出出错的消息也方便重新投递有助于排查线上问题。
3.4 延时队列
有时候我们会延时执行任务的需求即在过一段时间之后再执行任务。 AMQP 协议RabbitMQ 原生并没有直接支持延时队列的功能目前使用 RabbitMQ 来实现延时队列有两种方式
通过**TTL 和死信队列**结合来实现延时队列通过 RabbitMQ 的**x-delay-message-exchange 插件**来实现延时队列
3.4.1 TTL 死信队列实现延时队列
通过**TTL 和死信队列**结合来实现延时队列
首先声明一个队列队列上绑定死信队列并设置队列的超时时间不设置任何消费者监听此队列生产者将消息发送到队列上由于没有消费者监听队列则发送到队列中的消息经过超时时间之后就会转发到死信队列上由死信队列完成相应的延时处理业务逻辑 此种方法存在时序性的弊端因为队列的特殊性FIFO在同一个队列消息过期时间不同的情况下可能会出现先到期的后被执行延时逻辑的可能性。
3.4.2 x-delay-message-exchange 插件实现延时队列
通过 RabbitMQ 的**x-delay-message-exchange 插件**来实现延时队列
broker 集成 x-delay-message-exchange 插件使用 Docker 进行安装
docker pull feeld/x-delayed-rabbitmq声明延时队列交换机并指定交换机类型为 x-delayed-message可以通过指定 x-delayed-type 来指定实际交换机的类型将队列绑定到该延时队列交换机上发送消息时通过设置消息头参数 x-delay 或者 MessageProperties.setDelay() 方法可以设置延时时间单位ms如果两者都设置则以最后一次设置的时间为准
4. RabbitMQ 如何解决问题
这里说的问题即前面提到的使用消息队列的劣势中举例的一些问题
4.1 使用 RabbitMQ 集群实现高可用
RabbitMQ 有两种集群模式
普通集群模式镜像集群模式
4.2 如何保证消息不被重复消费
消息重复消费即同一条消息被消费者的消费逻辑处理了多遍即消息重复重复消费的问题。可能出现的原因如下
消息路由方式选择错误错误地使用了消息分发路由模式使本该路由给单个消费者的消息分发给了多个消费者每个消费者都执行了一次消息消费的逻辑。消息确认机制没有正常工作当出现网络波动或其他情况使生产者/消费者与 broker 之间的消息确认机制没有正常工作时则可能出现消息重复消费的问题。 这里我们只讨论消息确认机制没有正常工作的处理方法
在消息中增加一个 ID 字段用来标识消息的唯一性在消费者消费消息后将收到的消息的 ID 在存储层中进行查询 如果能查询出来则说明被处理过即消息已经被消费过执行比较更新或者其他逻辑如果查询不出来则说明没有被处理过即消息还没有被消费过则执行正常的处理逻辑并将这条消费记录存在存储层中
4.3 如何确保消息不丢失
从消息投递时序图我们可以知道消息的整个生命周期会经历生产者broker消费者三个终端则消息的防丢失需要从以下几个方面来进行考虑
消息从生产者到 broker 之间的防丢失消息在 broker 中的防丢失消息从 broker 到消费者之间的防丢失
4.3.1 消息从生产者到 broker 之间的防丢失
生产者将消息发送到 broker 时因为网络等原因消息可能会发送失败即出现了消息丢失。一般有两种方式来解决
RabbitMQ 提供的事务机制在 Spring AMQP 中可以通过设置
rabbitTemplate.setChannelTransacted(true);来启用事务。当消息没有正常到达 broker 时会抛出异常生产者可以捕获异常来感知到消息并没有正常到达 broker 从而执行重发等逻辑。但是缺点是开启事务后每个 channel 会采用同步阻塞的方式来等待事务的完成导致** broker 的吞吐量和性能会大幅下降**。
publishConfirm 机制 即发布确认模式 一旦使用了 publishConfirm 模式所有发布的消息都会被指派一个唯一的 ID一旦消息到达了 broker 之后broker 就会发送一个 ack 消息给生产者标识这个消息已经正常到达了 broker如果消息是持久化的那么确认消息会在消息写入磁盘之后发出。Spring AMQP 的相关配置见 publishConfirm 机制。相比于事务模式发布确认模式的性能比较优秀。
4.3.2 消息在 broker 中的防丢失
将消息、交换机、队列的都设置为持久化则可以避免消息在 broker 中的丢失。
消息持久化生产者发送消息时设置 deliveryMode 属性为 2 即可将消息设置为持久化在 Spring AMQP 中设置
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);进行消息持久化的设置
交换机/队列持久化在声明交换机/队列时设置 durable 为 true 即可。
4.3.3 消息从 broker 到消费者之间的防丢失
消息从 broker 到消费者之间的防丢失需要考虑以下两个方面
消息没有路由到队列由于消息不能匹配任何的队列路由逻辑在生产者设置了 mandatory 为 true 且开启了 returnCallback 配置时broker 将会把消息退回给生产者消息没有从队列到达消费者 在开启了队列持久化和消息持久化后如果消息和队列没设置过期时间即使 broker 宕机消息也不会丢失消息最终还是会发送到消费者手中在开启了队列持久化和消息持久化后如果消息和队列存在过期时间则在队列上配置死信队列将过期的消息发送到死信队列中进行处理 消息到达了消费者但是还没有被消费逻辑处理 将消息 ACK 改为手动确认模式在消息完成处理逻辑后手动发送 ACK 确认消费者收到消息如果消息处理逻辑中出现异常则将消息发往死信队列或者设置 requeue 属性为 true将消息发往其他队列
4.4 如何保证消息传递的顺序性
当消息有顺序状态时由于存在多个消费者则可能会出现消息最终处理顺序错乱的问题。
举例说明进入队列的消息是 123如果有多个消费者监听了这个队列那么由于网络或者自身处理速度等因素影响之后可能会出现消费者的处理顺序为 321 的情况解决办法
当消息需要强一致顺序时将具有顺序性的多条消息打包成一条消息发往 broker当消息不需要强一致顺序时可以在生产端给每一个消息增加一个自增 ID或者一个消息产生的时间戳用作消息顺序标记字段在消息消费的处理逻辑上进行消息顺序标记字段的判断 如果接收到的消息的顺序字段比当前处理过的消息当前顺序要大则正常消费如果接收到的消息的顺序字段比当前处理过的消息当前顺序要小则直接丢弃或者做一些其他逻辑处理
