什么是 MQ

MQ(message queue)从字面意思上看本质是个队列FIFO 先入先出只不过队列中存放的内容是
message 而已。
还是一种跨进程的通信机制用于上下游传递消息。
在互联网架构中MQ 是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。
使用了 MQ 之后消息发送上游只需要依赖 MQ不用依赖其他服务。

为什么要用 MQ

1.流量消峰

举个例子如果订单系统最多能处理一万次订单这个处理能力应付正常时段的下单时绰绰有余正
常时段我们下单一秒后就能返回结果。但是在高峰期如果有两万次下单操作系统是处理不了的只能限
制订单超过一万后不允许用户下单。使用消息队列做缓冲我们可以取消这个限制把一秒内下的订单分
散成一段时间来处理这时有些用户可能在下单十几秒后才能收到下单成功的操作但是比不能下单的体
验要好。

2.应用解耦

以电商应用为例应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后如果耦合
调用库存系统、物流系统、支付系统任何一个子系统出了故障都会造成下单操作异常。当转变成基于
消息队列的方式后系统间调用的问题会减少很多比如物流系统因为发生故障需要几分钟来修复。在
这几分钟的时间里物流系统要处理的内存被缓存在消息队列中用户的下单操作可以正常完成。当物流
系统恢复后继续处理订单信息即可中单用户感受不到物流系统的故障提升系统的可用性。

3.异步处理

有些服务间调用是异步的例如 A 调用 BB 需要花费很长时间执行但是 A 需要知道 B 什么时候可
以执行完。
以前一般有两种方式A 过一段时间去调用 B 的查询 api 查询。
或者 A 提供一个 callback apiB 执行完之后调用 api 通知 A 服务。
这两种方式都不是很优雅使用消息总线可以很方便解决这个问题
A 调用 B 服务后只需要监听 B 处理完成的消息当 B 处理完成后会发送一条消息给 MQMQ 会将此消息转发给 A 服务。
这样 A 服务既不用循环调用 B 的查询 api也不用提供 callback api。
同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息

MQ有什么缺点?

系统可用性降低
系统引入的外部依赖越多越容易挂掉本来你就是A系统调用BCD三个系统的接口就好了人 ABCD四个系统好好的没啥问题你偏加个MQ进来万一MQ挂了咋整MQ挂了整套系统崩溃了你不就完了么。(可以利用集群解决)

系统复杂性提高
硬生生加个MQ进来你怎么保证消息没有重复消费怎么处理消息丢失的情况怎么保证消息传递的顺序性头大头大问题一大堆痛苦不已。

一致性问题(保证消息不丢失)
A系统处理完了直接返回成功了人都以为你这个请求就成功了但是问题是要是BCD三个系统那里BD两个系统写库成功了结果C系统写库失败了咋整你这数据就不一致了。

所以消息队列实际是一种非常复杂的架构你引入它有很多好处但是也得针对它带来的坏处做各种额外的技术方案和架构来规避掉最好之后你会发现妈呀系统复杂度提升了一个数量级也许是复杂了10倍。但是关键时刻用还是得用的。

MQ 的分类

ActiveMQ

优点单机吞吐量万级时效性 ms 级可用性高基于主从架构实现高可用性消息可靠性较
低的概率丢失数据

缺点:官方社区现在对 ActiveMQ 5.x 维护越来越少高吞吐量场景较少使用。

Kafka

大数据的杀手锏谈到大数据领域内的消息传输则绕不开 Kafka这款为大数据而生的消息中间件
以其百万级 TPS 的吞吐量名声大噪迅速成为大数据领域的宠儿在数据采集、传输、存储的过程中发挥着举足轻重的作用。目前已经被 LinkedInUber, Twitter, Netflix 等大公司所采纳。

优点: 性能卓越单机写入 TPS 约在百万条/秒最大的优点就是吞吐量高。时效性 ms 级可用性非常高kafka 是分布式的一个数据多个副本少数机器宕机不会丢失数据不会导致不可用,消费者采用 Pull 方式获取消息, 消息有序, 通过控制能够保证所有消息被消费且仅被消费一次;有优秀的第三方
Kafka Web 管理界面 Kafka-Manager在日志领域比较成熟被多家公司和多个开源项目使用功能支持功能较为简单主要支持简单的 MQ 功能在大数据领域的实时计算以及日志采集被大规模使用

缺点Kafka 单机超过 64 个队列/分区Load 会发生明显的飙高现象队列越多load 越高发送消息响应时间变长使用短轮询方式实时性取决于轮询间隔时间消费失败不支持重试支持消息顺序但是一台代理宕机后就会产生消息乱序社区更新较慢

RocketMQ

RocketMQ 出自阿里巴巴的开源产品用 Java 语言实现在设计时参考了 Kafka并做出了自己的一
些改进。被阿里巴巴广泛应用在订单交易充值流计算消息推送日志流式处理binglog 分发等场景。

优点:单机吞吐量十万级,可用性非常高分布式架构,消息可以做到 0 丢失,MQ 功能较为完善还是分
布式的扩展性好,支持 10 亿级别的消息堆积不会因为堆积导致性能下降,源码是 java 我们可以自己阅
读源码定制自己公司的 MQ

缺点支持的客户端语言不多目前是 java 及 c++其中 c++不成熟社区活跃度一般,没有在 MQ
核心中去实现 JMS 等接口,有些系统要迁移需要修改大量代码

RabbitMQ

2007 年发布是一个在 AMQP(高级消息队列协议)基础上完成的可复用的企业消息系统是当前最
主流的消息中间件之一。

优点:由于 erlang 语言的高并发特性性能较好吞吐量到万级MQ 功能比较完备,健壮、稳定、易
用、跨平台、支持多种语言 如Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等支持 AJAX 文档齐全开源提供的管理界面非常棒用起来很好用,社区活跃度高更新频率相当高

缺点商业版需要收费,学习成本较高

MQ 的选择

Kafka

Kafka 主要特点是基于 Pull 的模式来处理消息消费追求高吞吐量一开始的目的就是用于日志收集
和传输适合产生大量数据的互联网服务的数据收集业务。大型公司建议可以选用如果有日志采集功能肯定是首选 kafka 了。

RocketMQ

天生为金融互联网领域而生对于可靠性要求很高的场景尤其是电商里面的订单扣款以及业务削
峰在大量交易涌入时后端可能无法及时处理的情况。RocketMQ 在稳定性上可能更值得信赖这些业务场景在阿里双 11 已经经历了多次考验如果你的业务有上述并发场景建议可以选择 RocketMQ。

RabbitMQ

结合 erlang 语言本身的并发优势性能好时效性微秒级社区活跃度也比较高管理界面用起来十分
方便如果你的数据量没有那么大中小型公司优先选择功能比较完备的 RabbitMQ。

RabbitMQ 的概念

RabbitMQ 是一个消息中间件它接受并转发消息。
你可以把它当做一个快递站点当你要发送一个包裹时你把你的包裹放到快递站快递员最终会把你的快递送到收件人那里按照这种逻辑 RabbitMQ 是一个快递站一个快递员帮你传递快件。
RabbitMQ 与快递站的主要区别在于它不处理快件而是接收存储和转发消息数据。

四大核心概念

生产者 Provider

产生数据发送消息的程序是生产者

消费者 Consumer

消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者消费
者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

交换机 Broker

交换机是 RabbitMQ 非常重要的一个部件一方面它接收来自生产者的消息另一方面它将消息
推送到队列中。
交换机必须确切知道如何处理它接收到的消息是将这些消息推送到特定队列还是推送到多个队列亦或者是把消息丢弃这个得有交换机类型决定

队列 Queue

队列是 RabbitMQ 内部使用的一种数据结构尽管消息流经 RabbitMQ 和应用程序但它们只能存
储在队列中。队列仅受主机的内存和磁盘限制的约束本质上是一个大的消息缓冲区。
许多生产者可以将消息发送到一个队列许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式

RabbitMQ基本概念

AMQP协议

AMQP协议所谓的高级消息队列协议可以把它理解成一种公认的协议规范就像http协议一样只是这个AMQP协议针对的是消息队列。这个协议使得遵从了它的规范的客户端应用和消息中间件服务器的全功能互操作成为可能。

Broker

接收和分发消息的应用RabbitMQ Server 就是 Message Broker

Virtual host

出于多租户和安全因素设计的把 AMQP 的基本组件划分到一个虚拟的分组中类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时可以划分出多个 vhost每个用户在自己的 vhost 创建 exchange／queue 等

Connection

publisher／consumer 和 broker 之间的 TCP 连接

Channel

如果每一次访问 RabbitMQ 都建立一个 Connection在消息量大的时候建立 TCP
Connection 的开销将是巨大的效率也较低。Channel 是在 connection 内部建立的逻辑连接如果应用程序支持多线程通常每个 thread 创建单独的 channel 进行通讯AMQP method 包含了 channel id 帮助客户端和 message broker 识别 channel所以 channel 之间是完全隔离的。Channel 作为轻量级的
Connection 极大减少了操作系统建立 TCP connection 的开销

Exchange

message 到达 broker 的第一站根据分发规则匹配查询表中的 routing key分发消息到 queue 中去。常用的类型有direct (point-to-point), topic (publish-subscribe) and fanout (multicast)

Queue

消息最终被送到这里等待 consumer 取走

Binding

exchange 和 queue 之间的虚拟连接binding 中可以包含 routing keyBinding 信息被保
存到 exchange 中的查询表中用于 message 的分发依据

vhost 是什么? 起什么作用?

每一个rabbitmq服务器都能创建虚拟的消息服务器我们称之为虚拟主机(virtual host)。简称vhost

特性
每一个vhost本质上是一个小型的独立的rabbitmq服务器拥有自己独立的完整的一套队列、绑定关系、交换器等。同一个服务器上的多个vhost是完全隔离的。队列及交换器等不互通。

所以一个broker可以开设多个vhost用于不同用户的权限分离

如何创建vhost?
1)通过前台页面的admin中创建
2)使用rabbitmqctl add_vhost vhost名称 命令

如何删除vhost?
1)前台删除
2)rabbitmqctl delete_vhost vhost_name

Rabbitmq里的交换机类型有哪些?

1、什么是Exchange

在RabbitMQ中生产者发送消息不会直接将消息投递到队列中而是先将消息投递到交换机中 在由交换机转发到具体的队列 队列再将消息以推送或者拉取方式给消费者进行消费.

2、路由键 ( RoutingKey)

生产者将消息发送给交换机的时候 会指定RoutingKey指定路由规则。

3、绑定键 ( BindingKey)

通过绑定键将交换机与队列关联起来 这样RabbitMQ就知道如何正确地将消息路由到队列。

4、关系

生产者将消息发送给哪个Exchange是需要由RoutingKey决定的.
生产者需要将Exchange与哪个队列绑定时需要由 BindingKey决定的。
当routing key和binding key能对应上的时候就发到相应的队列中。

交换机类型和区别

1、直连交换机 Direct exchange (点对点)

直连交换机的路由算法非常简单 将消息推送到binding key与该消息的routing key相同的队列。

直连交换机X上绑定了两个队列。第一个队列绑定了绑定键orange, 第二个队列有两个绑定键 black和green。

在这种场景下一 个消息在布时指定了路由键为orange将会只被路由到队列Q1
路由键为black 和green的消息都将被路由到队列Q2。其他的消息都将被丢失。

同一个绑定键可以绑定到不同的队列上去 可以增加一个交换机X与队列Q2的绑定键在这种清况下直连交换机将会和广播交换机有着相同的行为 将消息推送到所有匹配的队列。一个路由键为black的消息将会同时被推送到队列Q1和Q2。

它有一个特殊的属性使得它对于简单应用特别有用处 那就是每个新建队列 (queue) 都会自动绑定到默认交换机上绑定的 路由键(routing key) 名称与队列名称相同。

当你声明了一个名为“hello”的队列RabbitMQ会自动将其绑定到默认交换机上绑定binding的路由键名称也是为“hello”。

当携带着名为“hello”的路由键的信息被发送到默认交换机的时候此消息会被默认交换机路由至名为“hello”的队列中

类似amq.*的名称的交换机这些是RabbitMQ默认创建的交换机。

这些队列名称被预留做RabbitMQ内部使用不能被应用使用否则抛出403错误

2、主题交换机 Topic exchange (发布订阅)

直连交换机的缺点
直连交换机的 routing_key方案非常简单 如果我们希望一 条消息发送给多个队列 那么这个交换机需 要绑定上非常多的 routing_key.

假设每个交换机上都绑定一堆的 routing_key连接到各个队列上。那么消息的管理 就会异常地困难。

主题交换机的特点
发送到主题交换机的 消息不能有任意的 routing key, 必须是由点号分开的一串单词这些单词可以是任意的但通常是与消息相关的一些特征。

如以下是几个有效的routing key:

“stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabb 代”, routing key的单词可以 有很多最大限制是255 bytes。

Topic 交换机的 逻辑与 direct 交换机有点 相似 使用特定路由键发送的消息 将被发送到所有使用匹配绑定键绑定的队列 然而 绑定键有两个特殊的情况

*表示匹配任意一个单词

#表示匹配任意—个或多个单词

如

routing key quick.orange.rabbit-> queue Ql, Q2

routing key lazy.orange.elephant-> queue Ql,Q2

延申

当一个队列的绑定键是＂＃＂它将会接收所有的消息而不再考虑所接收消息的路由键。

当一个队列的绑定键没有用到"#"和’*"时它又像 direct 交换一样工作。

3、扇形交换机 Fanout exchange 广播

扇形交换机是最基本的交换机类型它所能做的事情非常简单广播消息。

扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。
因为广播不需要＇思考”所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

4、首部交换机 Headers exchange

类似主题交换机但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。
此交换机有个重要参数”x-match”

当”x-match”为“any”时消息头的任意一个值被匹配就可以满足条件
当”x-match”设置为“all”的时候就需要消息头的所有值都匹配成功

5、Dead Letter Exchange (死信交换机)

RabbitMQ作为一个高级消息中间件提出了死信交换器的概念。

这种交互器专门处理死了的信息被拒绝可以重新投递的信息不能算死的。

消息变成死信一般是以下三种情况

①、消息被拒绝并且设置requeue参数为false。

②、消息过期默认情况下Rabbit中的消息不过期但是可以设置队列的过期时间和信息的过期的效果

③、队列达到最大长度一般当设置了最大队列长度或大小并达到最大值时

当满足上面三种情况时消息会变成死信消息并通过死信交换机投递到相应的队列中。

我们只需要监听相应队列就可以对死信消息进行最后的处理。

订单超时处理

生产者生产一条1分钟后超时的订单信息到正常交换机exchange-a中消息匹配到队列queue-a但一分钟后仍未消费。
消息会被投递到死信交换机dlx-exchange中并发送到私信队列中。
死信队列dlx-queue的消费者拿到信息后根据消息去查询订单的状态如果仍然是未支付状态将订单状态更新为超时状态。

交换机的属性

Name:交换机名称

Type:交换机类型directtopicfanoutheaders

Durability:是否需要持久化如果持久性则RabbitMQ重启后交换机还存在

Auto Delete:当最后一个绑定到Exchange上的队列删除后自动删除该Exchange

Internal:当前Exchange是否用于RabbitMQ内部使用默认为false。

Arguments:扩展参数用于扩展AMQP协议定制使用

RabbitMQ的工作模式

1.简单模式

简单模式是最简单的消息模式它包含一个生产者、一个消费者和一个队列。生产者向队列里发送消息消费者从队列中获取消息并消费。

2.工作模式

工作模式是指向多个互相竞争的消费者发送消息的模式它包含一个生产者、两个消费者和一个队列。两个消费者同时绑定到一个队列上去当消费者获取消息处理耗时任务时空闲的消费者从队列中获取并消费消息。

3.发布/订阅模式

发布/订阅模式是指同时向多个消费者消费消息的模式类似广播的形式它包含一个生产者、两个消费者、两个队列和一个交换机。两个消费者同时绑定到不同的队列上去两个队列绑定到交换机上去生产者通过发送消息到交换机所有消费者接收并消费消息。

4.路由模式

路由模式是可以根据路由键选择性给多个消费者发送消息的模式它包含一个生产者、两个消费者、两个队列和一个交换机。两个消费者同时绑定到不同的队列上去两个队列通过绑定键绑定到交换机上去生产者发送消息到交换机交换机通过路由键转发到不同队列队列绑定的消费者接收并消费消息

(要求路由键和绑定键相同)
常用的交换器主要分为一下三种
fanout如果交换器收到消息将会广播到所有绑定的队列上
direct如果路由键完全匹配消息就被投递到相应的队列
topic可以使来自不同源头的消息能够到达同一个队列。 使用 topic 交换器时可以使用通配符

5. 通配符模式

就是绑定键是某种泛泛的符号规则,如果传过来的消息路由键与绑定键匹配,就能传递到对应的消息队列上
特殊匹配符号
*只能匹配一个单词
#可以匹配零个或多个单词。

RabbitMQ工作流程

生产者消息投递过程

生产者连接到Broker 建立一个连接然后开启一个信道

接着生产者声明一个交换器 并设置相关属性比如交换机类型、是否持久化、是否自动删除、是否内置等

生产者声明一个队列并设置相关属性比如是否排他、是否持久化、是否自动删除、消息最大过期时间、消息最大长度、消息最大字节数等

生产者通过绑定键将交换器和队列绑定起来

生产者发送消息至Broker 发送的消息包含消息体和路由规则(可以理解为路由键)、交换器、优先级、是否持久化、过期时间、延时时间等信息的标签

相应的交换器根据接收到的路由键和与队列的绑定键查找相匹配的队列如果找到 则将从生产者发送过来的消息存入相应的队列中

如果没有找到 则根据生产者配置的属性选择丢弃还是回退给生产者

关闭信道

关闭连接

消费者消费消息过程

消费者连接到Broker 建立一个连接开启一个信道

消费者向 RabbitMQ Broker 请求消费相应队列中的消息在这个过程中可能会设置消费者标签、是否自动确认、是否排他等

等待 RabbitMQ Broker 回应并投递相应队列中的消息 消费者接收消息。

消费者确认接收到的消息

RabbitMQ从队列中删除相应己经被确认的消息

关闭信道

关闭连接。

涉及名词解释

在上方的消息流转过程中涉及了以下几个名词

是否持久化

将数据持久化到磁盘中

是否自动删除

当一个队列或交换机的所有消费者都与之断开连接时则这个队列或交换机就会自动删除

是否内置

客户端程序无法直接发送消息到这个交换器中只能通过交换器路由到交换器这种方式

是否排他

如果一个队列被声明为排他队列该队列仅对首次声明它的连接可见并在连接断开时自动删除。这里需要注意的是

排他队列是基于连接可见的同一个连接的不同信道是可以同时访问同一连接创建的排他队列; "首次"是指如果一个连接己经声明了排他队列其他连接是不允许建立同名的排他队列的这个与普通队列不同:即使该队列是持久化的一旦连接关闭或者客户端退出该排他队列都会被自动删除这种队列适用于一个客户端同时发送和读取消息的应用场景。

自动确认

消费者在订阅队列时可以指定 autoAck 参数当 autoAck 等于 false时 RabbitMQ会等待消费者显式地回复确认信号后才从内存(或者磁盘)中移去消息(实质上是先打上删除标记之后再删除)

当 autoAck 等于 true 时 RabbitMQ 会自动把发送出去的消息置为确认然后从内存(或者磁盘)中删除而不管消费者是否真正地消费到了这些消息

采用消息确认机制后只要设置 autoAck 参数为 false 消费者就有足够的时间处理消息不用担心处理消息过程中消费者进程挂掉后消息丢失的问题。因为 RabbitMQ 会一直等待持有消息直到消费者显式确认收到消息

说一下消息确认机制和返回机制,有什么区别?

为了保证 RabbitMQ 中消息的可靠性投递以及消息在发生特定异常时的补偿策略,RabbitMQ诞生了消息确认和返回机制.

消息确认机制(也叫应答机制)

消息确认机制是保障消息与 RabbitMQ消息之间可靠传输消息一种保障机制
其主要内容就是用来监听RabbitMQ消息队列收到消息之后返回给生产端的ack确认消息
消息确认机制描述了一种消息是否已经被发送到 RabbitMQ消息队列中以及 RabbitMQ消息队列是否以及接收到生产端发送的消息。

消息确认机制的作用
监听生产者的消息是否已经发送到了 RabbitMQ消息队列中
如果消息没有被发送到 RabbitMQ消息队列中则消息确认机制不会给生产端返回任何确认应答也就是没有发送成功
相反如果消息被成功发送到了 RabbitMQ消息队列中则消息确认机制会给生产端返回一个确认应答以通知生产者消息已经发送到了 RabbitMQ消息队列

消息返回机制

描述不可达的消息与生产者之间的一种保障策略
其主要是用来监听RabbitMQ消息队列中是否存在不可达的消息并根据监听结果返回给生产端的一种监听机制
消息返回机制描述了一种 RabbitMQ消息队列中的不可达消息与生产端的关系

什么是不可达的消息
消息在被成功发送到RabbitMQ消息队列中之后如果消息在经过当前配置的 exchangeName 或 routingKey 没有找到指定的交换机或没有匹配到对应的消息队列
那么这个消息就被称为不可达的消息如果此时配置了消息返回机制那么此时RabbitMQ消息队列会返回给生产者一个信号信号中包括消息不可达的原因以及消息本身的内容。

channel.addReturnListener(new ReturnListener() {
    @Override
    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.out.println("return relyCode: " + replyCode);    // 状态码
        System.out.println("return replyText: " + replyText);   // 结果信息
        System.out.println("return exchange: " + exchange);
        System.out.println("return routingKey: " + routingKey);
        System.out.println("return properties: " + properties);
        System.out.println("return body: " + new String(body));

    }
});

Rabbitmq的持久化机制

RabbitMQ 的消息默认存放在内存上面如果不特别声明设置消息不会持久化保存到硬盘上面的如果节点重启或者意外crash掉消息就会丢失。所以就要对消息进行持久化处理。

rabbitmq的持久化分为队列持久化、消息持久化和交换器持久化。

1. 队列的持久化是在定义队列时的durable参数来决定的当durable为true时才代表队列会持久化。

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//第二个餐胡设置为true代表队列持久化
channel.queueDeclare("queue.persistent.name", true, false, false, null);

2. 如果要在重启后保持消息的持久化必须设置消息是持久化的标识。

//通过传入MessageProperties.PERSISTENT_PLAIN就可以实现消息持久化
channel.basicPublish("exchange.persistent", "persistent", MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent_test_message".getBytes());

3. 上面阐述了队列的持久化和消息的持久化如果不设置exchange的持久化对消息的可靠性来说没有什么影响但是同样如果exchange不设置持久化那么当broker服务重启之后exchange将不复存在那么既而发送方rabbitmq producer就无法正常发送消息。这里建议同样设置exchange的持久化。exchange的持久化设置也特别简单.

所以一般三个持久化设置都要进行.

可以对所有消息都持久化吗?
不可以
持久化的操作是将数据写入磁盘中,效率上肯定比写入内存中要慢很多倍.而我们一般用mq会处理很多业务消息,若是所有消息都持久化,压力无疑是巨大的.所以持久化策略需要综合考虑,以及可能遇到的问题和解决方案,或者我们可以让一些必要数据持久化.

Rabbitmq的应答机制

一、消息应答机制

消费者完成一个任务可能需要一段时间如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉由于RabbitMQ 一旦向消费者传递了一条消息便立即将该消息标记为删除。在这种情况下突然有个消费者挂掉了我们将丢失正在处理的消息以及后续发送给该消费者的消息也无法接收到。

为了保证消息在发送过程中不丢失rabbitmq 引入消息应答机制。

消费者在接收到消息并且处理该消息之后告诉 rabbitmq 它已经处理了rabbitmq 可以把该消息删除掉了。

二、自动应答

消息发送后立即被认为已经传送成功这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前消费者出现连接或者 channel 关闭那么消息就丢失了,当然另一方面这种模式消费者可以传递过载的消息没有对传递的消息数量进行限制当然这样有可能使得消费者由于接收太多还来不及处理的消息导致这些消息的积压最终使得内存耗尽最终这些消费者线程被操作系统杀死所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

三、手动消息应答的方法

//用于肯定确认RabbitMQ 已知道该消息并且成功的处理消息可以将其丢弃了
channel.basicAck();

//用于否定确认
channel.basicNack();

//用于否定确认,不处理该消息了直接拒绝可以将其丢弃了
channel.basicReject();

如果消费者由于某些原因失去连接(其通道已关闭连接已关闭或 TCP 连接丢失)导致消息未发送 ACK 确认RabbitMQ 将了解到消息未完全处理并将对其重新排队。如果此时其他消费者可以处理它将很快将其重新分发给另一个消费者。这样即使某个消费者偶尔死亡也可以确保不会丢失任何消息。

Rabbitmq事务机制

1、概述
在使用RabbitMQ的时候我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失除此之外我们还会遇到一个问题当消息的发布者在将消息发送出去之后消息到底有没有正确到达broker代理服务器呢如果不进行特殊配置的话默认情况下发布操作是不会返回任何信息给生产者的也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的如果在消息到达broker之前已经丢失的话持久化操作也解决不了这个问题因为消息根本就没到达代理服务器你怎么进行持久化那么这个问题该怎么解决呢

RabbitMQ为我们提供了两种方式

通过AMQP事务机制实现这也是AMQP协议层面提供的解决方案
通过将channel设置成confirm模式来实现
2、事务机制
这里首先探讨下RabbitMQ事务机制。

RabbitMQ中与事务机制有关的方法有三个txSelect(), txCommit()以及txRollback(), txSelect用于将当前channel设置成transaction模式txCommit用于提交事务txRollback用于回滚事务在通过txSelect开启事务之后我们便可以发布消息给broker代理服务器了如果txCommit提交成功了则消息一定到达了broker了如果在txCommit执行之前broker异常崩溃或者由于其他原因抛出异常这个时候我们便可以捕获异常通过txRollback回滚事务了。

public class P1 {

    private static final String QUEUE_NAME = "test_tx";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,true,null);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd 'at' HH:mm:ss z");
        Date date = new Date(System.currentTimeMillis());
        String message = simpleDateFormat.format(date);

        try {
            channel.txSelect();//开始事务
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            channel.txCommit();//提交事务
        }catch (Exception e){
            channel.txRollback();//回滚事务
            System.out.println("send message txRollback");
        }
        channel.close();
        connection.close();


    }
}

消息分发策略有哪些?

1、轮询分发

RabbitMQ 分发消息默认采用的轮询分发但是在某种场景下这种策略并不是很好当有两个消费者在处理任务时其中有个消费者 处理任务的速度非常快而另外一个消费者处理速度却很慢这个时候我们还是采用轮询分发就会导致这处理速度快的这个消费者很大一部分时间处于空闲状态。我们可以通过修改消息分发的默认机制来达到优化目的

2、不公平分发

通过设置参数 channel.basicQos(1);实现不公平分发策略使得能者多劳

通过RabbitMq的Web管理页面可以看到Channels的Prefetch count属性显示为1则表示不公平分发成功

上面介绍了basicQos如果我们将qos的值设为1那么你想一想会出现什么情况呢信道中只允许传输一条消息那么当这条消息处理完后队列会立马发送下一条消息所以这个时候快的不断处理慢的等待当前处理完再处理下一条。这样就实现了能者多劳。

3、预值分发

当消息被消费者接收后但是没有确认此时这里就存在一个未确认的消息缓冲区用于存储非被确认的消息该缓存区的大小是没有限制的。

预取值 定义通道上允许的未确认消息的最大数量。一旦未确认消息数量达到配置的最大数量RabbitMQ 将停止在通道上传递更多消息除非至少有一个未处理的消息被确认例如假设在通道上有未确认的消息 5、6、78并且通道的预取值计数设置为 4此时 RabbitMQ 将不会在该通道上再传递任何消息除非至少有一个未应答的消息被 ack例如 tag=6 这个消息刚刚被确认 ACK此时RabbitMQ 将会感知这个情况到并再发送一条消息。

如果消费者消费了大量的消息但是没有确认的话就会导致消费者连接节点的内存消耗变大所以找到合适的预取值是一个反复试验的过程不同的负载该值取值也不同 100 到 300 范 围内的值通常可提供最佳的吞吐量。

其实预取值设置就是一个非公平分发策略,我们通过设置qos来设置消息缓冲区允许未确认消息的最大数量,当我们设置为1时,是最保守的,吞吐量也是最低的,而100到300这个范围通常是最佳的.

如何保证消息的顺序消费

消息在投入到queue的时候是有顺序如果只是单个消费者来处理对应的单个queue是不会出现消息错乱的问题。但是在消费的时候有可能多个消费者消费同一个queue由于各个消费者处理消息的时间不同导致消息未能按照预期的顺序处理。其实根本的问题就是如何保证消息按照预期的顺序处理完成。

出现消费顺序错乱的情况

1. 为了提高处理效率一个queue存在多个consumer

2. 一个queue只存在一个consumer但是为了提高处理效率consumer中使用了多线程进行处理

保证消息顺序性的方法
将原来的一个queue拆分成多个queue每个queue都有一个自己的consumer。该种方案的核心是生产者在投递消息的时候根据业务数据关键值例如订单ID哈希值对订单队列数取模来将需要保证先后顺序的同一类数据同一个订单的数据 发送到同一个queue当中

一个queue就一个consumer在consumer中维护多个内存队列根据业务数据关键值例如订单ID哈希值对内存队列数取模将消息加入到不同的内存队列中然后多个真正负责处理消息的线程去各自对应的内存队列当中获取消息进行消费。

总结
核心思路就是根据业务数据关键值划分成多个消息集合而且每个消息集合中的消息数据都是有序的每个消息集合有自己独立的一个consumer。多个消息集合的存在保证了消息消费的效率每个有序的消息集合对应单个的consumer也保证了消息消费时的有序性。

消息队列如何保证消息的可靠性传输

消息的可靠性传输分为两个问题,一个是保证消息不被重复消费,另一个是保证消息不丢失
保证消息不重复被消费,就是保证消息的幂等性问题,消息的幂等性是指一个操作执行任意多次所产生的影响均与一次执行的影响相同,在mq里,也就是消息只能被消费一次,不能被重复消费.

消息丢失如何处理:
发送方丢失,可能发送方在发送消息的过程中,出现网络问题等导致mq接收不到消息,导致了消息丢失.
要解决这个问题,首先可以采用事务机制,在发送消息的时候实现事务机制,若是出现发送失败的情况,可以进行回滚,而让消息重新被发送.但是开启了事务,发送方就必须同步等待事务执行完毕或者回滚,导致消息一多,性能会下降.

但是,还有一个更好的办法:可以采用确认机制,发送方在发送消息的时候必须要保证要收到一个确认消息,如果没有收到或者收到失败的确认消息,就说明消息发送失败,要重新进行发送,确认机制是可以采用异步进行的,这样就极大地保证了在保留效率的基础上又能保证消息的不丢失问题.

第二个丢失问题可能是在mq方发生的,如果mq没有进行持久化,出现了宕机关机等情况,消息就会丢失,解决办法无非就是将消息进行持久化,这样在出现问题的时候可以及时对消息进行恢复.

第三个丢失问题可能在消费方发生,这和发送方丢失问题类似,解决这个问题也是采用确认机制,这样一来就可以实现效率上的保证和消息不丢失的保证
.
消息重复如何处理
但是解决了这些问题,就会产生下面的幂等性问题:
我们都知道mq是可以进行重发的,且只有在它认为失败的情况会进行重发.什么时候mq会认为它发送给消费者的消息是失败的呢?也就是超出了它等待消费者响应的时间,这是一个超时时间,若是过了这个时间消费者仍然没有响应,说明mq发送失败,就会进行重试,而其实这个时候消费者可能是没有失败的,它只是因为某个原因导致消费超出了mq的等待时间而已,这个时候mq再发送一次消息,消费者就会重复消费.
实现幂等性消费:
① 通过数据库比如处理订单时记录订单ID在消费前去数据库中进行查询该记录是否存在如果存在则直接返回。
② 使用全局唯一ID再配合第三组主键做消费记录比如使用 redis 的 set 结构生产者发送消息时给消息分配一个全局ID在每次消费者开始消费前先去redis中查询有没有消费记录如果消费过则不进行处理如果没消费过则进行处理消费完之后就将这个ID以k-v的形式存入redis中(过期时间根据具体情况设置)。

如何处理消息堆积情况

场景题几千万条数据在MQ里积压了七八个小时。

出现该问题的原因

消息堆积往往是生产者的生产速度与消费者的消费速度不匹配导致的。有可能就是消费者消费能力弱渐渐地消息就积压了也有可能是因为消息消费失败反复复重试造成的也有可能是消费端出了问题导致不消费了或者消费极其慢。比如消费端每次消费之后要写mysql结果mysql挂了消费端hang住了不动了或者消费者本地依赖的一个东西挂了导致消费者挂了。

所以如果是 bug 则处理 bug如果是因为本身消费能力较弱则优化消费逻辑比如优化前是一条一条消息消费处理的那么就可以批量处理进行优化。

临时扩容快速处理积压的消息

1先修复 consumer 的问题确保其恢复消费速度然后将现有的 consumer 都停掉

2临时创建原先 N 倍数量的 queue 然后写一个临时分发数据的消费者程序将该程序部署上去消费队列中积压的数据消费之后不做任何耗时处理直接均匀轮询写入临时建立好的 N 倍数量的 queue 中

3接着临时征用 N 倍的机器来部署 consumer每个 consumer 消费一个临时 queue 的数据

4等快速消费完积压数据之后恢复原先部署架构 重新用原先的 consumer 机器消费消息。

这种做法相当于临时将 queue 资源和 consumer 资源扩大 N 倍以正常 N 倍速度消费。

MQ长时间未处理导致MQ写满的情况如何处理

如果消息积压在MQ里并且长时间都没处理掉导致MQ都快写满了这种情况肯定是临时扩容方案执行太慢这种时候只好采用 “丢弃+批量重导” 的方式来解决了。首先临时写个程序连接到mq里面消费数据消费一个丢弃一个快速消费掉积压的消息降低MQ的压力然后在流量低峰期时去手动查询重导丢失的这部分数据

短时间内无法扩容或者扩容无法完全解决问题

可以尝试降低一些非核心业务的消息处理,其次可以通过监控排查,优化消费者端的业务代码或者查看是否存在一些消息被重复消息的情况.

如何保证消息队列的高可用

RabbitMQ 是基于主从非分布式做高可用性的。

RabbitMQ 有三种模式单机模式、普通集群模式、镜像集群模式

单机模式
一般没人生产用单机模式

普通集群模式

普通集群模式用于提高系统的吞吐量通过添加节点来线性扩展消息队列的吞吐量。
也就是在多台机器上启动多个 RabbitMQ 实例而队列 queue 的消息只会存放在其中一个 RabbitMQ 实例上但是每个实例都同步 queue 的元数据元数据是 queue 的一些配置信息通过元数据可以找到 queue 所在实例。消费的时候如果连接到了另外的实例那么该实例就会从数据实际所在的实例上的queue拉取消息过来就是说让集群中多个节点来服务某个 queue 的读写操作

但普通集群模式的缺点在于无高可用性queue所在的节点宕机了其他实例就无法从那个实例拉取数据RabbitMQ 内部也会产生大量的数据传输。

镜像队列集群模式

镜像队列集群是RabbitMQ 真正的高可用模式集群中一般会包含一个主节点master和若干个从节点slave如果master由于某种原因失效那么按照slave加入的时间排序"资历最老"的slave会被提升为新的master。

镜像队列下所有的消息只会向master发送再由master将命令的执行结果广播给slave所以master与slave节点的状态是相同的。比如每次写消息到 queue 时master会自动将消息同步到各个slave实例的queue如果消费者与slave建立连接并进行订阅消费其实质上也是从master上获取消息只不过看似是从slave上消费而已比如消费者与slave建立了TCP连接并执行Basic.Get的操作那么也是由slave将Basic.Get请求发往master再由master准备好数据返回给slave最后由slave投递给消费者。

从上面可以看出队列的元数据和消息会存在于多个实例上也就是说每个 RabbitMQ 节点都有这个 queue 的完整镜像任何一个机器宕机了其它机器节点还包含了这个 queue 的完整数据其他消费者都可以到其它节点上去消费数据。

1缺点

① 性能开销大消息需要同步到所有机器上导致网络带宽压力和消耗很重

② 非分布式没有扩展性如果 queue 的数据量大到这个机器上的容量无法容纳了此时该方案就会出现问题了

2如何开启镜像集群模式呢

在RabbitMQ 的管理控制台Admin页面下新增一个镜像集群模式的策略指定的时候是可以要求数据同步到所有节点的也可以要求同步到指定数量的节点再次创建 queue 的时候应用这个策略就会自动将数据同步到其他的节点上去了。

谈谈你对死信队列的理解

概念
当queue消息队列中的消息由于一些原因没办法被消费如果这些消息一直没办法被处理就会从这个正常的消息队列转移到死信消息队列中。

应用场景
当用户下单之后如果一直不支付那么用户下单的这条消息就会被存放到死信队列中去。

死信的来源
1.当queue消息队列满的时候再进来的消息就会被放到死信队列中去。

2.在消息应答的时候如果消费者一直没有告诉RabbitMQ有没有成功处理消息那么RabbitMQ消息队列就不清楚自己到底要不要删除这条消息这个时候消息队列中的消息一直没办法处理这样这条消息也会被放到死信队列中去。

3.消息TTL过期什么意思TTL的全拼是Time To Live意思是指存活时间就是消息队列中存放的消息一般都是有一定时间的超过了这个时间这条消息就会被放到死信队列中去。

如何去实现死信队列?
生产者:
发送消息.

普通消费者:
1)在用Map声明普通队列参数的时候,可以先设置死信队列的路由key和交换机,让普通该队列处理不来的消息转到对应的死信队列上.
2)然后声明普通队列,普通交换机,死信队列,死信交换机.
3)之后消费普通队列消息

public class Consumer01 {
    //普通交换机的名称
    public static final String NORMAL_EXCHANGE="normal_exchange";
    //死信交换机的名称
    public static final String DEAD_EXCHANGE="dead_exchange";
    //普通队列的名称
    public static final String NORMAL_QUEUE="normal_queue";
    //死信队列的名称
    public static final String DEAD_QUEUE="dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();

        //声明死信和普通交换机类型为direct类型
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        //声明普通队列
        Map<String,Object> arguments=new HashMap<>();
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        arguments.put("x-dead-letter-routing-key","lisi");
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);

        //声明死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);

        //绑定普通的交换机与普通的队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        //绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //如果能成功接收到消息会调用的回调函数
        DeliverCallback deliverCallback=(consumerTag, message)->{
            System.out.println("Consumer01接收者接收到的消息:"+new String(message.getBody()));
        };

        //如果取消从消息队列中获取消息时会调用的回调函数
        CancelCallback cancelCallback= consumerTag->{
            System.out.println("消息消费被中断");
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
    }
}

死信消费者:
1)消费死信队列即可,消费过程和普通队列一样.