RocketMQ 入门教程
共 7800字，需浏览 16分钟
·
2021-03-05 19:41
本文公众号来源：大鱼仙人
作者：大鱼
本文已收录至我的GitHub
消息队列RocketMQ版是阿里云基于Apache RocketMQ构建的低延迟、高并发、高可用、高可靠的分布式消息中间件。
所谓黄河之水天上来，大自然间每一个事物都不是平白无故来的吧？🐒🐒怎么来的，🐒🐒它母亲生产的；香奈儿💄💄怎么来的，机器加原料生产的；就连平时吃的大米，也是有出处的；咱们是怎么来的，咱们当然是伟大的母亲生产下来的了
顺便感谢一下伟大的母亲，周日记得给她打个电话哦
下面进入主题，这是分割线
消息队列RocketMQ版既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。下面列举了一些特点
消息查询：消息队列RocketMQ版提供了三种消息查询的方式，分别是按Message ID、Message Key以及Topic查询
查询消息轨迹：通过消息轨迹，能清晰定位消息从生产者发出，经由消息队列RocketMQ版服务端，投递给消息消费者的完整链路，方便定位排查问题
集群消费和广播消费：当使用集群消费模式时，消息队列RocketMQ版认为任意一条消息只需要被消费者集群内的任意一个消费者处理即可；当使用广播消费模式时，消息队列RocketMQ版会将每条消息推送给消费者集群内所有注册过的消费者，保证消息至少被每台机器消费一次
重置消费位点：根据时间或位点重置消费进度，允许用户进行消息回溯或者丢弃堆积消息
死信队列：将无法正常消费的消息储存到特殊的死信队列供后续处理
全球信息路由：用于全球不同地域之间的消息同步，保证地域之间的数据一致性
客户端，其实很容易理解了，我们可以把RocketMQ理解成一个消息服务，既然是一个服务，我们就需要调用这个服务，那么调用这个服务的时候，这个消息从哪里来，这个就是要根据业务场景来定了，所以啊，消息的生产者Producer属于一个客户端；消息产生了，总不能一直放着吧，总要有人处理掉这些消息吧，这也是业务决定的，所以消息的消费者consumer也是属于客户端。
下面啊，三歪就带着大家一起来看看这客户端的用处
生产者Producer，顾名思义，就是负责生产消息的，此时大家应该脑子有很多问号才对，比如Producer发消息发到哪里了，流程是怎么样的，发的消息都是什么类型的等等这些，这些问题搞懂了的话，Producer这个客户端基本就搞定了
三歪教大家一个小技巧，学习一个东西，先搞懂大体流程，再拆分而细攻之，最后再统筹理解，这样效果会很好，独家秘方
接下来我从消息是如何发送的（负载均衡、容错机制）、消息发给谁和存储到哪里、消息的类型三方面来介绍Producer
1、消息是如何发送的？
首先，消息总不能产生了哪里也不去吧，那产生这个消息就没有任何意义了，所以这个消息总要发送到一个地方去，接力传递，看下面这个图
Producer会首先从本地缓存中获取到指定的Topic，如果找到就直接根据这个Topic发送产生的消息，缓存大家都明白啊，就是为了优化速度，减少网络传
没有的话，就要去NameServer获取最新的Topic列表（这个是Broker启动的时候注册到NameServer上的），通过一定的策略选择一个MessageQueue队列，获取这个mq所在的Broker地址，也是先从本地缓存中获取，如果获取不到则请求NameServer获取（NameServer中也同样注册了Broker地址和Topic的映射关系），进行发送消息
发送失败的话，会有重试机制，默认是重试三次
其实保存这么多，既能减少和NameServer之间的网络传输，又能减小NameServer的压力，NameServer本身就是属于轻量级的设计，这样也有利于减轻NameServer的压力，NameServer我也会单独写一篇来介绍
负载均衡
我们知道消息发送的时候会首先选择一个对应的Topic，每个Topic会对应多个MessageQueue，这样就有一个问题，发消息的时候要是做不到雨露均沾，可能就会有的队列多，有的队列少这样的问题，就会造成资源的浪费
RocketMQ采用了朴素的方式，没错，就是轮询，高端的食材往往只需要最朴素的烹饪方式~
生产者通过轮询某个 Topic 下的所有 MessageQueue 的方式来实现发送方的负载均衡，简单来说就是人人都有份，如下图：
通过这种方式，可以将一个 Topic 的消息分散到多个 MessageQueue 上，进而分散到多个 Broker 上。
发送消息的容错机制：
Producer 作为发送消息的一方，有3种容错机制：
本地缓存：把从 NameSever 获取的信息缓存到本地，以防 NameSever 宕机
不可用Broker集合：Producer有一个 Broker 的容错机制，开关
sendLatencyFaultEnable可以开启，RocketMq内部会维护一个故障Broker的HashMap，把一定延迟级别的Broker放入这个map，下次选择Broker的时候，就会规避不可用的Broker。重试：Producer发送消息时，有一个重试机制，默认重试3次。死信队列 Consumer消费重试超过指定次数，进入死信队列
通过这种方式，可以将一个 Topic 的消息分散到多个 MessageQueue 上，进而分散到多个 Broker 上。
2、消息发给谁和存储在哪里？
Producer连接NameSever
Producer 通过 NameSever 获取指定 Topic 的 Broker 路由信息，并在本地保存一份缓存数据，比如一个Topic有哪些 MessageQueue，MessageQueue 在哪几台 Broker 上，Broker 的ip.port等等。Producer 发送消息只发到 Master Broker上，Slave 通过主从同步获取数据。
那么 Produce 是怎么连接NameSever 的呢
连接：单个生产者者和一台 Nameserver 保持长连接，定时查询topic配置信息，如果该nameserver挂掉，生产者会自动连接下一个nameserver，直到有可用连接为止，并能自动重连。
轮询时间：默认情况下，生产者每隔30秒从nameserver获取所有topic的最新队列情况，这意味着某个broker如果宕机，生产者最多要30秒才能感知，在此期间，发往该broker的消息发送失败。该时间由DefaultMQProducer的pollNameServerInteval参数决定，可手动配置。
心跳：与nameserver没有心跳
Producer连接Broker
连接：生产者 跟 Topic 涉及的所有Broker 保持长连接。
心跳：默认情况下，生产者每隔30秒向所有broker发送心跳。broker每隔10秒钟（此时间无法更改），扫描所有还存活的连接，若某个连接2分钟内（当前时间与最后更新时间差值超过2分钟，此时间无法更改）没有发送心跳数据，则关闭连接
Producer连接上Broker之后，消息会通过轮询的方式发送到Broker上，并且存储在Broker中的CommitLog中，这里面存储的是原始消息，还有一个ConsumeQueue用于存储投递到某一个queue的消息的位置信息。当然，消息队列会持久化到磁盘中的，不影响内存，当然也会定期清理消息。
如果再面试的时候，你不仅能说出RocketMQ的用处，你还能说出它的存储原理和寻址原理，那面试官就爱上你了。此时你再拿出王炸，就是解决各种实际问题的能力，比如如何处理重复消息啊、如何保证消息的顺序性啊、在分布式系统中如何保证分布式事务啊
面试官当场给你发offer，say：How much money do you expect to work for us ？
3、消息的种类
RocketMQ种的消息种类大致可以分为四种：普通消息、定时和延时消息、顺序消息、事务消息四种类型，这是重点！
简单介绍下四种类型
普通消息：消息队列RocketMQ版中无特性的消息，区别于有特性的定时和延时消息、顺序消息和事务消息。
定时和延时消息：允许消息生产者对指定消息进行定时（延时）投递，最长支持40天。
顺序消息：允许消息消费者按照消息发送的顺序对消息进行消费。
事务消息：实现类似X或Open XA的分布事务功能，以达到事务最终一致性状态。
消息队列RocketMQ提供的四种消息类型所对应的Topic不能混用，例如，创建的普通消息的Topic只能用于收发普通消息，不能用于收发其他类型的消息；同理，事务消息的Topic也只能收发事务消息，不能用于收发其他类型的消息，以此类推
普通消息：消息队列RocketMQ中无特性的消息，区别于有特性的定时和延时消息、顺序消息和事务消息
普通消息以三种发送方式：同步Sync发送、异步Async发送和单向Oneway发送
同步就是我们发送了消息之后必须等到服务器响应之后才能发送下一个；异步适用于对时间较敏感的业务场景，异步不需要等待服务器的响应就可以连续发送消息；单向则比异步用时更短，一般在微秒级别，但是可靠性会降低，因为只管发送，不等待服务器响应，也没有回调函数触发
同步发送
同步，消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式
异步发送
异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式
消息队列RocketMQ版的异步发送，需要实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应，并处理响应结果
一般用于对时间较敏感的业务场景
单向发送
发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别
应用于对可靠性要求并不高的场景，比如日志收集
定时和延时消息：允许消息生产者对指定消息进行定时（延时）投递，最长支持40天
延时消息用于指定消息发送到消息队列RocketMQ版的服务端后，延时一段时间才被投递到客户端进行消费（例如3秒后才被消费），适用于解决一些消息生产和消费有时间窗口要求的场景，或者通过消息触发延迟任务的场景，类似于延迟队列。
定时消息可以做到在指定时间戳之后才可被消费者消费，适用于对消息生产和消费有时间窗口要求，或者利用消息触发定时任务的场景。
适用场景
通过消息来触发一些定时任务，这个时候这个定时消息就派上用场了，比如在某一时间向用户发送的提醒消息；一些消息生产和消费之间有时间窗口，比如典型的电商里面的超时未支付关闭订单的场景，这时延时消息就派上用场了，超时未完成支付就关闭订单
定时消息的精度会有1s~2s的延迟误差
其实定时消息和延时消息在使用的时候也是有一些差别的，用过的应该都知道，给大家提一下，定时消息需要明确指定消息发送时间点之后的某一时间点作为消息投递的时间点；延时消息则需要设定一个延时的时间长度，长度是固定的，但是时刻点不是固定，是根据发送消息的时间点有关的，消息将从当前发送时间点开始延迟固定时间之后才开始投递，这个大家应该都很清楚了，淘宝下个单，给你留30分钟时间支付，超时未支付则关闭订单
顺序消息：允许消息消费者按照消息发送的顺序进行消息的发送
顺序消息分为两类：
全局顺序：对于指定的一个Topic，所有消息按照严格的先入先出FIFO（First In First Out）的顺序进行发布和消费。
分区顺序：对于指定的一个Topic，所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段，和普通消息的Key是完全不同的概念。
其实这也是个比较经典的问题，面试也是比较常问的，就是如何保证顺序性？三歪反正会回答，你会吗？
不会的出来挨打~~
如果遇到这个问题，首先你要分情况说明，就是分为全局顺序和分区顺序这两种情况：
1、全局顺序适用于性能要求不高，所有的消息都要严格按照先进先出的顺序来发布和消费的场景。这种情况我也没遇到过，一般也不太会使用全局有序这种
2、分区顺序适用于性能要求比较高，以Sharding Key作为分区字段，在用一个区块中严格按照先进先出的顺序发布和消费。比如用户注册的时候的验证码，以用户ID作为Sharding Key，那么同一个用户发送的消息都会按照发布的先后顺序来消费，再比如就是电商中的订单流程问题
阿里巴巴集团内部电商系统均使用分区顺序消息，既保证业务的顺序，同时又能保证业务的高性能。别问我怎么知道的，阿里云官网写的
顺序消息常见问题
为什么全局顺序消息性能一般？
全局顺序消息是严格按照FIFO的消息阻塞原则，即上一条消息没有被成功消费，那么下一条消息会一直被存储到Topic队列中。如果想提高全局顺序消息的TPS，可以升级实例配置，同时消息客户端应用尽量减少处理本地业务逻辑的耗时。
顺序消息支持哪种消息发送方式？是否支持集群消费和广播消费？
顺序消息只支持可靠同步发送方式，不支持异步发送方式，否则将无法严格保证顺序。顺序消息暂时仅支持集群消费模式，不支持广播消费模式。
事务消息：实现类似X或者Open XA的分布式事务功能，以达到最终一致性
消息队列RocketMQ版提供类似X或Open XA的分布式事务功能，通过消息队列RocketMQ版事务消息，能达到分布式事务的最终一致。
半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列RocketMQ版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。
消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列RocketMQ版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或是Rollback），该询问过程即消息回查。
跟小仙来看看事务消息发送步骤：
1、发送方将半事务消息发送到服务端Broker，服务端会将消息持久化，成功之后会返回ACK确认消息已经发送成功，此时消息为半事务消息
2、发送方开始执行本地事务的逻辑
3、发送方会根据本地事务的执行结果向服务端提交二次确认，决定Commit还是Rollback，服务端收到Commit之后则把这个消息标记为可投递，发送到消费方；服务端收到Rollback之后则删除半事务消息，服务端不会发送，则消费方也不会收到
如可是如果断网或者应用重启这些情况，上述的步骤的二次确认信息无法到达服务端，怎么办？
这里其实有个回查机制，发送方发送消息之后，需要本地执行事务，如果事务执行的过程出现卡死的情况，或者事务执行结果因为网络等问题，无法传递事务结果到服务端，服务端会执行一个回查机制，来确认这个半事务消息的最终提交情况
消息队列RocketMQ版的消费者和生产者客户端对象是线程安全的，可以在多个线程之间共享使用。可以在服务器上（或者多台服务器）部署多个生产者和消费者实例，也可以在同一个生产者或消费者实例里采用多线程发送或接收消息，从而提高消息发送或接收TPS。避免为每个线程创建一个客户端实例。
好了，回顾一下本篇的内容吧
1、消息发送的负载均衡、容错机制
2、消息发送流程和存储
3、消息的类型：普通消息（同步发送、异步发送、单向发送）、定时和延时消息、顺序消息（全局顺序和部分顺序）、事务消息
《对线面试官》系列目前已经连载16篇啦！进度是一周更新两篇，欢迎持续关注
点击小卡片关注【面试造火箭】
关注后回复「888」还可获取【面试资料】网盘地址哟！