模块十八_RocketMQ

序言:

文章内容输出来源：拉勾教育Java高薪训练营。
本篇文章是学习课程中的一部分课后笔记

一、 RocketMQ介绍、概念

1. RocketMQ介绍

• RocketMQ在阿里内部叫做Metaq（最早名为Metamorphosis，中文意思“变形记”，是作家卡夫卡的中篇小说代表作，可见是为了致敬Kafka）。

• RocketMQ是Metaq 3.0之后开源的版本。

• Metaq最早源于Kafka，早期借鉴了Kafka很多优秀的设计。但是由于Kafka是Scale语言编写而阿里系主要使用Java，且无法满足阿里的电商、金融业务场景，所以誓嘉（花名）团队用Java重新造轮子，并做了大量的改造和优化。
  在此之前，淘宝有一款消息中间件名为Notify，目前已经逐步被Metaq所取代。

• 第一代的Notify主要使用了推模型，解决了事务消息；第二代的MetaQ主要使用了拉模型，解决了顺序消息和海量堆积的问题。相比起Kafka使用的Scale语言编写，RabbitMQ 使用Erlang语言编写，基于Java的RocketMQ开源后更容易被广泛的研究，以及其他大厂定制开发。

rocketMq介绍.png

2. RocketMQ的使用场景

• 应用解耦
  系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。
• 流量削峰
  应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。
  举例:业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显然不划算，这时可以使用消息队列对峰值流量削峰
• 数据分发
  通过消息队列可以让数据在多个系统之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可

二、 RocketMQ特性

1. RocketMQ特性

• 1 订阅与发布
  消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息。

• 2 消息顺序
  消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以并行消费的。RocketMQ可以严格的保证消息有序。

• 3 消息过滤
  RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在Broker端实现的，优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担、而且实现相对复杂。

• 4 消息可靠性
  RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：
  ① Broker非正常关闭
  ② Broker异常Crash
  ③ OS Crash
  ④ 机器掉电，但是能立即恢复供电情况
  ⑤ 机器无法开机（可能是cpu、主板、内存等关键设备损坏）
  ⑥ 磁盘设备损坏
  ①、②、③、④ 四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。
  ⑤、⑥属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。
  RocketMQ在这两种情况下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息可能丢失。
  通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与Money相关的应用。注：RocketMQ从3.0版本开始支持同步双写。

• 5 至少一次
  至少一次(At least Once)指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。

• 6 回溯消费
  回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒。

• 7 事务消息
  RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。
  RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致性。

• 8 定时消息
  定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level。
  messageDelayLevel是broker的属性，不属于某个topic。发消息时，设置delayLevel等级即可：msg.setDelayLevel(level)。level有以下三种情况：

  • level == 0，消息为非延迟消息
  • 1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
  • level > maxLevel，则level== maxLevel，例如level==20，延迟2h

定时消息会暂存在名为SCHEDULE_TOPIC_XXXX的topic中，并根据delayTimeLevel存入特定的queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟的消息能够顺序消费。broker会调度地消费SCHEDULE_TOPIC_XXXX，将消息写入真实的topic。

需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都会变高。

• 9 消息重试
  Consumer消费消息失败后，要提供一种重试机制，令消息再消费一次。Consumer消费消息失败，通常可以认为有以下几种情况：

  • 1. 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其它消息，而这条失败的消息即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10秒后再重试。
  • 2. 由于依赖的下游应用服务不可用，例如db连接不可用，外系统网络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用sleep 30s，再消费下一条消息，这样可以减轻Broker重试消息的压力。

• 10 消息重投
  生产者在发送消息时：
  1、同步消息失败会重投
  2、异步消息有重试
  3、oneway没有任何保证。
  消息重投保证消息尽可能发送成功、不丢失，但可能会造成消息重复，消息重复在RocketMQ中是无法避免的问题。
  消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会是大概率事件。另外，生产者主动重发、consumer负载变化也会导致重复消息。

如下方法可以设置消息重试策略：

• 1. retryTimesWhenSendFailed：同步发送失败重投次数，默认为2，因此生产者会最多尝试发送retryTimesWhenSendFailed + 1次。不会选择上次失败的broker，尝试向其他broker发送，最大程度保证消息不丢失。超过重投次数，抛异常，由客户端保证消息不丢失。当出现RemotingException、MQClientException和部分MQBrokerException时会重投。
• 2. retryTimesWhenSendAsyncFailed：异步发送失败重试次数，异步重试不会选择其他broker，仅在同一个broker上做重试，不保证消息不丢。
• 3. retryAnotherBrokerWhenNotStoreOK：消息刷盘（主或备）超时或slave不可用（返回状态非SEND_OK），是否尝试发送到其他broker，默认false。十分重要消息可以开启。

• 11 流量控制
  生产者流控，因为broker处理能力达到瓶颈；消费者流控，因为消费能力达到瓶颈。

  • 1. 生产者流控：
       commitLog文件被锁时间超过osPageCacheBusyTimeOutMills时，参数默认为1000ms，发生流控。
       如果开启transientStorePoolEnable = true，且broker为异步刷盘的主机，且
       transientStorePool中资源不足，拒绝当前send请求，发生流控。
       broker每隔10ms检查send请求队列头部请求的等待时间，如果超过
       waitTimeMillsInSendQueue，默认200ms，拒绝当前send请求，发生流控。
       broker通过拒绝send 请求方式实现流量控制。
       注意，生产者流控，不会尝试消息重投。
  • 2. 消费者流控：
       消费者本地缓存消息数超过pullThresholdForQueue时，默认1000。
       消费者本地缓存消息大小超过pullThresholdSizeForQueue时，默认100MB。
       消费者本地缓存消息跨度超过consumeConcurrentlyMaxSpan时，默认2000。
       消费者流控的结果是降低拉取频率。

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

  RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），
  将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。
  在RocketMQ中，可以通过使用console控制台对死信队列中的消息进行重发来使得消费者实例再次进行消费。

2.消费模式Push or Pull

RocketMQ消息订阅有两种模式，一种是Push模式（MQPushConsumer），即MQServer主动向消费端推送；另外一种是Pull模式（MQPullConsumer），即消费端在需要时，主动到MQ Server拉取。
但在具体实现时，Push和Pull模式本质都是采用消费端主动拉取的方式，即consumer轮询从broker拉取消息。

• Push模式:
  好处就是实时性高。不好处在于消费端的处理能力有限，当瞬间推送很多消息给消费端时，容易造成消费端的消息积压，严重时会压垮客户端。
• Pull模式
  好处就是主动权掌握在消费端自己手中，根据自己的处理能力量力而行。缺点就是如何控制Pull的频率。定时间隔太久担心影响时效性，间隔太短担心做太多“无用功”浪费资源。比较折中的办法就是长轮询。
• Push模式与Pull模式的区别:
  Push方式里，consumer把长轮询的动作封装了，并注册MessageListener监听器，取到消息后，唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。
  Pull方式里，取消息的过程需要用户自己主动调用，首先通过打算消费的Topic拿到
  MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue。

RocketMQ使用长轮询机制来模拟Push效果，算是兼顾了二者的优点。