RocketMQ高级应用

消息中间件对比

MQ - 图1

为什么选择RocketMQ

性价比,社区活跃度
性价比之“性”:
性能:阿里支撑,经受住淘宝,天猫双11重重考验;性能高;可靠性好;可用性高;易扩展
功能:功能完善,我们需要的功能,基本都够满足,如:事务消息,消息重试,死信队列,定时消息等;
易用,跨平台:跨语言,多协议接入(支持HTTP, MQTT, TCP协议,支持Restful风格HTTP收发消息)
性价比之“价”:
钱能解决的问题,一般都不是问题,所以免费服务不能满足的,适当的花钱购买所需服务是值得的,早期阿里引进的IOE,那我们引进的就是RocketMQ的阿里云和VIP服务;
当然,钱解决不了的问题,那必然是问题,IOE的高消费,不如去IOE寻找技术方面的突破,
社区活跃度:
技术突破就分能突破的和不能突破的,需要社区支持,社区不活跃,版本问题不改善,自己修复问题不仅耗时,而且未必正确能够经受考验

RocketMQ队列消费谨记

一个消费者可以消费多个队列,但一个队列只能由一个消费者消费

RocketMQ消息顺序和重复消费问题

RocketMQ特性

顺序性
消息过滤
消息持久化
消息回溯
大量消息堆积
定时消息
消息重试

RocketMQ广播与集群区别

RocketMQ高可用

RocketMQ分布式事务

RocketMQ有哪些坑

RocketMQ的namesrv全挂掉是否影响通信

RocketMQ的生产者和发送者Reblance

RocketMQ队列中的消息有序,能否保证消费者消费也有序

必须是顺序消费才可以,并行消费无法做到有序。是否受网络影响,顺序消费也有可能无序,待测

RocketMQ怎么设置消息是从Master消费还是从Slaver消费。Master和Slaver同时在线,消息是否会从Master消费一遍,然后再从Slaver消费一遍?

RocketMQ默认端口
namesrv默认端口9876,broker默认端口10911,VIP默认端口10909,每个broker启动后默认占用两个端口10911和10913或10912和10914

RocketMQ同一台机,启动多个生产者实例或多个消费者实例,需要设置不同的实例名称

RocketMQ无论何种情况(发送到单个队列,顺序消费或并行消费;或者发送到多个队列,顺序消费或者并行消费),延时/定时消息总是迟于非延时/非定时消息到达broker,延时/定时消息是在延时/定时时间过后才被投放到broker,也即延时/定时消息不会影响非延时/非定时消息到达broker。如果等待所有消息全部到达broker之后,才启动消费者,这个延时/定时是否还有意义?

RocketMQ在不考虑网络影响的情况下,只有将同一ID的消息发送到同一队列上,并且消费端使用顺序消费,才能保证消息被顺序消费。即生产者不使用MessageQueueSelector或者消费者不使用MessageListenerOrderly的任何一种情况出现都不能保证消息被顺序消费,考虑网络影响的情况待测。

RocketMQ消费端消息ack是ack到本地队列,然后由本地队列登记后,再5秒钟间隔上报到broker?

经测试:调大上报时间,本地ack之后(返回CONSUME_SUCCESS,且#Diff > 0),立即停掉消费端,过许久,发现仍#Diff > 0,证明消费端确实还没有上报ack进度给broker;重启消费端之后,因#Diff > 0,消息又被重新消费了一次,证明之前broker确实没收到ack,也由此可证明消费端ack是先ack到本地队列,停掉消费端,本地队列的所有信息都没有了,也就因此迟迟不会将#Diff变为0。

什么是解耦?
系统的耦合性越高,容错性就越低,以电商应用为例,用户创建订单后,如果耦合调用库存系统,物流系统,支付系统,任何一个子系统出了故障或者因为升级等原因暂时不可用,都会造成下单操作异常,影响用户使用体验。、

使用消息队列解耦合,系统的耦合性就会提高了,比如物流系统发生故障,需要几分钟才能修复,这段时间 内,物理系统要处理的数据被缓存 到消息队列中,用户的下单操作正常完成,当物流系统恢复后,补充处理存在消息队列中的订单消息即可,终端系统感知不到物流系统发生过几分钟故障。
image.png
什么是异步?
举个例子吧,对于生产者来说,生产者生产消息通过namesrv找到相应的broker路由,然后又broker路由按照某种策略(随机或者是轮训),选择一个队里发送消息,发送消息到队列后,返回发送结构,对于同步消息来说,我们需要等待发送消息结构返回后,才能继续执行,这个过程是耗时的,对于异步消息来说,当消息发送数据之后,采用消息回调的方式获取返回的结果,在这个过程中,线程不需要等待,可以继续执行其他的代码。

我在这里截取一下同步生产和异步生产的代码,做对比感受一下
image.png
什么是削峰填谷?
应用系统如果遇到系统请求流量的瞬间猛增,有可能会系统压垮。有了消息队列可以将大量的请求缓存起来,分散到很长一段时间处理,这样可以大大提高系统的稳定性和用户体验
image.png
一般情况,为了保证系统的稳定性,如果系统负载超过阈值,就会阻止用户请求,这会影响用户体验,而如果使用消息队列将请求缓存起来,等待系统处理完毕后通知用户下单完毕,这样总比不能下单体验要好。

另外处于经济考量的目的,业务系统正常时段的qps如果是1000,流量最高峰是10000,为了应对流量最高峰配置高性能的服务器显然是不划算的,这时可以使用消息队列对峰值流量削峰。


RocketMQ 执行流程
image.png
集群的工作流程

启动NameServer,NameServer起来后监听端口,等待Broker,Producer,Consumer连接上。相当于是一个路由控制中心。
Broker启动,跟所有的NameServer保持长连接,定时发送心跳包,心跳包中包含当前Broker信息(IP+端口等)以及存储所有topic信息,注册成功后,NameServer集群中就有topic和broker的映射关系。
发送消息前,先创建topic,创建topic是需要指定topic要存储在哪些Broker上,也可在发送消息的时候自动创建topic
Producer发送消息,启动时先和nameServer集群中的一台建立长连接,并从NameServer中获取当前的topic存放在哪个broker上面,轮询从队列列表中选择一个队列,然后与队列所在的Broker建立长连接从而向Broker发消息。
Consumer和Producer类似,跟其中一个NameServer建立长连接,获取当前订阅的topic在哪个broker上面,然后直接跟Broker建立连接通道,开始消费消息。
nameServer维护的路由信息如下:
image.png
怎么理解 Producer 的?
Producer即消费的生产者,消息的发送者发送消息的步骤主要是,

创建消息的生产者,并制定生产者组名
制定NameServer的地址
启动producer
创建消息对象,指定主题Topic,Tag和消息体
发送消息处理结果
关闭producer
可以发送同步消息(直接返回消息发送结果),可以发送异步消息(通过方法回调的方式,获取结果),可以发送单向消息(没有任何返回结果),我们支持发送消息的类型也是多样的

发送顺序消息
发送延时消息
发送批量消息
发送过滤消息
发送事务消息
发送顺序消息,消息有序指的是可以按照消息的发送顺序来消费(FIF0),RocketMQ可以严格的保证消息有序,可以分为分区有序或者全局有序。

顺序消费的原理解析,在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列);而消费消息的时候从多个queue上拉取消息,这种情况发送和消费是不能保证顺序拉取,但是如果控制发送的顺序消息只依次发送到同一个queue中,消费的时候只从这个queue上依次拉取,则就保证了顺序

例如一个订单的顺序流程是:创建,付款,推送,完成。订单号相同的消息会被先后发送到同一个队列中,消费是,同一个OrderId获取到的肯定是同一队列。
延时消息,比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单状态,如果还没有付款,就取消这个订单释放库存。
image.png
现在RocketMQ并不支持任意时间的延迟,需要设置几个固定的延时等级,从1s到2h分别对应着等级1到18
还有批量消息 ,过滤消息和事务消息,这里我就不一一说了,大家可以去参考相关资料。

怎么理解 Consumer 的?
consumer就是消息的消费者,消息的消费有两种模式,一种是集群模式(负载均衡模式),还有一种是广播模式。

集群模式下消费,每条消息只需要投递到订阅这个topic的Consumer Group下的一个实例即可。个queue只分给一个consumer实例,一个consumer实例可以允许同时分到不同的queue。但是如果consumer实例的数量比message queue的总数量还多的话,多出来的consumer实例将无法分到queue,也就无法消费到消息,也就无法起到分摊负载的作用了。所以需要控制让queue的总数量大于等于consumer的数量
广播模式
由于广播模式下要求一条消息需要投递到一个消费组下面所有的消费者实例,所以也就没有消息被分摊消费的说法。
在实现上,其中一个不同就是在consumer分配queue的时候,所有consumer都分到所有的queue

消费者消费模式有哪几种?
消费者获取消息有几种模式?
定时消息是什么?怎样实现的?
RocketMQ 如何保证高可用的?
RocketMQ分布式集群是通过Master和Slave的配合达到高可用性的。

Master和Slave的区别:在Broker的配置文件中,参数 brokerId的值为0表明这个Broker是Master,大于0表明这个Broker是 Slave,同时brokerRole参数也会说明这个Broker是Master还是Slave。

Master角色的Broker支持读和写,Slave角色的Broker仅支持读,也就是 Producer只能和Master角色的Broker连接写入消息;Consumer可以连接 Master角色的Broker,也可以连接Slave角色的Broker来读取消息。

消费者消费高可用

在Consumer的配置文件中,并不需要设置是从Master读还是从Slave 读,当Master不可用或者繁忙的时候,Consumer会被自动切换到从Slave 读。有了自动切换Consumer这种机制,当一个Master角色的机器出现故障后,Consumer仍然可以从Slave读取消息,不影响Consumer程序。这就达到了消费端的高可用性。

生产者生产高可用

在创建Topic的时候,把Topic的多个Message Queue创建在多个Broker组上(相同Broker名称,不同 brokerId的机器组成一个Broker组),这样当一个Broker组的Master不可 用后,其他组的Master仍然可用,Producer仍然可以发送消息。 RocketMQ目前还不支持把Slave自动转成Master,如果机器资源不足, 需要把Slave转成Master,则要手动停止Slave角色的Broker,更改配置文 件,用新的配置文件启动Broker。
消息主从复制

同步复制

同步复制方式是等Master和Slave均写成功后才反馈给客户端 写成功状态;
在同步复制方式下,如果Master出故障,slave上有全部的备份数据,容易恢复,但是同步复制会增大数据写入延迟,降低系统吞吐量

异步复制

异步复制方式是只要Master写成功 即可反馈给客户端写成功状态。
在异步复制方式下,系统拥有较低的延迟和较高的吞吐量,但是如果Master出了故障,有些数据因为没有被写 入Slave,有可能会丢失;
实际应用中要结合业务场景,合理设置刷盘方式和主从复制方式, 尤其是SYNC_FLUSH方式,由于频繁地触发磁盘写动作,会明显降低 性能。通常情况下,应该把Master和Save配置成ASYNC_FLUSH的刷盘 方式,主从之间配置成SYNC_MASTER的复制方式,这样即使有一台 机器出故障,仍然能保证数据不丢,是个不错的选择。


如何保证消息不被重复消费?或者说如何保证消息消费时的幂等性?
在互联网应用中,尤其在网络不稳定的情况下,消息队列 RocketMQ 的消息有可能会出现重复,这个重复简单可以概括为以下情况:

发送时消息重复

当一条消息已被成功发送到服务端并完成持久化,此时出现了网络闪断或者客户端宕机,导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息,消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

投递时消息重复

消息消费的场景下,消息已投递到消费者并完成业务处理,当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次,消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息,消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

负载均衡时消息重复(包括但不限于网络抖动、Broker 重启以及订阅方应用重启)

当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时,会触发 Rebalance,此时消费者可能会收到重复消息。

因为 Message ID 有可能出现冲突(重复)的情况,所以真正安全的幂等处理,不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据,而业务的唯一标识可以通过消息 Key 进行设置


如何保证消息的可靠性传输?要是消息丢失了怎么办?
对于生产者来说,可以向多个master的broker去发送消息,同时可以发送同步消息和异步消息返回服务方消息的应答,保证消息是否发送成功,对于消息,有同步刷盘和异步刷盘机制,主从之间也有同步复制和异步复制,保证了消息不丢失。同时我们也可以把消息记录的日志文件或者表中,RocketMQ消息的存储是有ConsumerQueue和ConmmitLog配合完成,消息真正的物理存储文件是CommitLog,ConsumerQueue是消息的逻辑队列,类似于数据库的索引文件,存储的是指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的ConsumerQueue文件,通过刷盘和复制机制来保证数据的高可用。

对于消费者来说,消费者既可以消费broker的master,有可以消费broker的slave,当master宕机之后,会自动切换的slave进行消费。对于广播的消息来说,我们可以进行消息的重试,消息队列RocketMQ默认允许每条消息最多重试16次,一条消息无论重试多少次,这些重试消息的MessageID不会改变。

当一条消息初次消费失败,消息队列RocketMQ会自动进行消息重试,达到最大重试次数后,若消费依然失败,则表明消费者在正常情况下无法正确消费该消息,此时,消息队列RocketMQ不会立刻将消息丢弃,而是将其发送到该消费者对应的特殊队列中。

死信消息具有以下特性

不会再被消费者正常消费。

有效期与正常消息相同,均为 3 天,3 天后会被自动删除。因此,请在死信消息产生后的 3 天内及时处理。

死信队列具有以下特性:

一个死信队列对应一个 Group ID, 而不是对应单个消费者实例。

如果一个 Group ID 未产生死信消息,消息队列 RocketMQ 不会为其创建相应的死信队列。

一个死信队列包含了对应 Group ID 产生的所有死信消息,不论该消息属于哪个 Topic。


如何保证消息的顺序性?
如何解决消息队列的延时以及过期失效问题?
消息队列满了以后该怎么处理?有几百万消息持续积压几小时,说说怎么解决?
大量消息在 mq 里积压了几个小时了还没解决
几千万条数据在 MQ 里积压了七八个小时,从下午 4 点多,积压到了晚上 11 点多。这个是我们真实遇到过的一个场景,确实是线上故障了,这个时候要不然就是修复 consumer 的问题,让它恢复消费速度,然后傻傻的等待几个小时消费完毕。这个肯定不能在面试的时候说吧。

一个消费者一秒是 1000 条,一秒 3 个消费者是 3000 条,一分钟就是 18 万条。所以如果你积压了几百万到上千万的数据,即使消费者恢复了,也需要大概 1 小时的时间才能恢复过来。

一般这个时候,只能临时紧急扩容了,具体操作步骤和思路如下:

先修复 consumer 的问题,确保其恢复消费速度,然后将现有 consumer 都停掉。
新建一个 topic,broker是原来的 10 倍,临时建立好原先 10 倍的 queue 数量。
然后写一个临时的分发数据的 consumer 程序,这个程序部署上去消费积压的数据,消费之后不做耗时的处理,直接均匀轮询写入临时建立好的 10 倍数量的 queue。
接着临时征用 10 倍的机器来部署 consumer,每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍,以正常的 10 倍速度来消费数据。
等快速消费完积压数据之后,得恢复原先部署的架构,重新用原先的 consumer 机器来消费消息。


如何解决高性能读写数据的问题?
总结起来就两点,顺序读写,零拷贝

磁盘如果使用得当,磁盘的速度完全可以匹配上网络 的数据传输速度。目前的高性能磁盘,顺序写速度可以达到600MB/s, 超过了一般网卡的传输速度。但是磁盘随机写的速度只有大概100KB/s,和顺序写的性能相差6000倍!因为有如此巨大的速度差别,好的消息队列系统会比普通的消息队列系统速度快多个数量级。RocketMQ的消息用顺序写,保证了消息存储的速度。

2)消息发送

Linux操作系统分为【用户态】和【内核态】,文件操作、网络操作需要涉及这两种形态的切换,免不了进行数据复制。

一台服务器 把本机磁盘文件的内容发送到客户端,一般分为两个步骤:

1)read;读取本地文件内容;

2)write;将读取的内容通过网络发送出去。

这两个看似简单的操作,实际进行了4 次数据复制,分别是:

从磁盘复制数据到内核态内存;

从内核态内存复 制到用户态内存;

然后从用户态 内存复制到网络驱动的内核态内存;

最后是从网络驱动的内核态内存复 制到网卡中进行传输。
通过使用mmap的方式,可以省去向用户态的内存复制,提高速度。这种机制在Java中是通过MappedByteBuffer实现的

RocketMQ充分利用了上述特性,也就是所谓的“零拷贝”技术,提高消息存盘和网络发送的速度。

这里需要注意的是,采用MappedByteBuffer这种内存映射的方式有几个限制,其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存,这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因12了


单机 RocketMQ 的 QPS 上限是多少?
对于单机的RocketMQ的吞吐量可以到达10万的数量级