rocketmq 消息指定_进大厂必备的RocketMQ你会吗？

点击关注"故里学Java"

右上角"设为星标"好文章不错过

关于消息队列，相信大家都不陌生，现在的中大型项目中或多或少都有使用到消息队列，对于消息队列大家可能都有一定的了解，使用消息队列可以解决什么样的问题，又会带来哪些问题相信也有了解，在前边的文章《消息队列(一)》中有介绍，感兴趣的小伙伴可以点击文章名直接打开。

今天我们主要介绍一下RocketMQ，关于RocketMQ很多人只知道是阿里开源的一款MQ中间件，实际工作中还是用的RabblitMQ，本文以及接下来几篇文章，我会分享一下RocketMQ相关的知识，详细的介绍一下RocketMQ，希望可以帮助到需要的朋友们。

RocketMQ基本概念

Producer：消息生产者，负责生产消息，一般由业务系统负责生产消息，消息生产者把业务应用产生的消息发送到broker服务器，
Consumer：消息消费者，从broker服务器拉取消息或者接收broker推送的消息进行消费处理。
Broker：消息中转的角色，负责存储消息、转发消息。
Name Server：充当路由消息的提供者，生产者或消费者可以通过名字查找所需broker的IP列表，集群部署的时候，各个NameServer实例是相互独立的，没有信息交换。
Topic：表示一类消息的集合，是RocketMQ进行消息订阅的基本单位。
Message Queue：用于存储消息的物理地址，每个Topic中是消息地址存储于多个Message Queue中。
Message：消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。

RocketMQ的特性

RocketMQ的功能特性很多，这里只简单介绍几个：

顺序消息

指的是可以按照消息的发送顺序来消费，有时候一组消息需要按照指定的顺序消费才有意义，但是多个消费者是并行消费的，RocketMQ可以严格的保证消息的有序。顺序消息可以分为两类：全局顺序消息和分区顺序消息。全局顺序是指某个topic下所有的消息都要保证顺序，部分顺序消息只要保证每一组消息被顺序消费。

全局顺序消费：对于指定的一个Topic，所有的消息按照严格的先入先出的顺序依次进行发布和消费
分区顺序消费：对于指定的一个Topic，所有的消息根据sharding key进行区块分区，同一个分区的消息按照先入先出的顺序进行消费。性能较全局顺序高。

事务消息

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

定时消息

定时消息是指消息发送到broker后，不会立即消费，等到设定的设定的实际才会投递给真正的topic。broker有配置项messageDelayLevel，默认值有1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也可以配置自定义的messageDelayLevel，在发送消息的时候，设置delayLevel等级即可：msg.setDelayLevel(level)。

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

回溯消息

回溯消息是指消费者以及消费成功，由于业务需要，需要重新消费，如果要实现此功能，Broker在向消费者投递成功消息后，消息仍然需要保留。重新消费一般是有一定时间纬度的，RocketMQ支持按照时间回溯消息，时间维度可以精确到毫秒。

消息重投

生产者在发送消息时，同步消费失败会重投，异步消息有重试，oneway没有任何保证。消息重投可以最大限度的保证消息发送成功、不丢失，但是也会导致消息重复，当消息量大、网络不好的时候消息重复的概率就会提高。

我们可以根据需要设置消息重试策略：

retryTimesWhenSendFailed：同步发送失败重投次数，默认为2 ，在重投的时候不会发送给上次失败的broker，会尝试向其他的broker发送，尽可能的保证消息不丢失。
retryTimesWhenSendAsyncFailed：异步发送失败重试次数，异步的重试还是选择上次的broker，不会选择其他的broker，不能保证消息不丢失。
retryAnotherBrokerWhenNotStoreOK：消费刷盘超时或者slave不可用，是否尝试发送给其他的broker，默认为false，非常重要的消息我们可以开启。

死信队列

死信队列用于处理消费失败的消息，当消息消费失败的时候，会自动进行消息重试，如果达到最大重试次数后，还是没有消费成功，就说明正常情况下不能正确的消费该消息，此时消息队列会把这个消息发送到该消费者对应的特殊队列中。RocketMQ将这种消息称为死信消息，将这种存储死信消息的队列称为死信队列，可以通过console控制台对死信队列中的消息进行重发。

流量控制

生产者流控，一般是因为broker处理能力达到了上限。消费者流控，一般是因为消费者消费能力达到了上限。

生产者流控：

commitLog文件被锁时间超过osPageCacheBusyTimeOutMills时，参数默认为1000ms，返回流控。
如果开启transientStorePoolEnable == true，且broker为异步刷盘的主机，且transientStorePool中资源不足，拒绝当前send请求，返回流控。
broker每隔10ms检查send请求队列头部请求的等待时间，如果超过waitTimeMillsInSendQueue，默认200ms，拒绝当前send请求，返回流控。
broker通过拒绝send 请求方式实现流量控制。

生产者流控不会尝试消息重投。

消费者流控：

消费者本地缓存消息数超过pullThresholdForQueue时，默认1000。
消费者本地缓存消息大小超过pullThresholdSizeForQueue时，默认100MB。
消费者本地缓存消息跨度超过consumeConcurrentlyMaxSpan时，默认2000。

消费者流控会降低拉取频率。

RocketMQ技术架构

以多Master多Slave模式为例：

我们可以看到，RocketMQ架构上分为四部分，分别为Producer、Consumer、NameServer、Broker。

Producer：消息发布的角色，支持分布式集群方式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。
Consumer：消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。
NameServer：NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。
Broker：Broker主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker包含了以下几个重要子模块。

RocketMQ部署架构

以多Master多Slave模式为例：

集群的工作流程;

启动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建立连接通道，开始消费消息。