RocketMQ面面观

不念不忘少年蓝@ 2023-06-30 01:52 10阅读 0赞

# RocketMQ面面观 #

## Rocket基本概念 ##

--------------------

### 1 消息模型（Message Model） ###

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。

### 2 消息生产者（Producer） ###

负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

### 3 消息消费者（Consumer） ###

负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。

### 4 主题（Topic） ###

表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。

### 5 代理服务器（Broker Server） ###

消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。

### 6 名字服务（Name Server） ###

名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。

### 7 拉取式消费（Pull Consumer） ###

Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。

### 8 推动式消费（Push Consumer） ###

Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高。

### 9 生产者组（Producer Group） ###

同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。

### 10 消费者组（Consumer Group） ###

同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。

### 11 集群消费（Clustering） ###

集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。

### 12 广播消费（Broadcasting） ###

广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。

### 13 普通顺序消息（Normal Ordered Message） ###

普通顺序消费模式下，消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。

### 14 严格顺序消息（Strictly Ordered Message） ###

严格顺序消息模式下，消费者收到的所有消息均是有顺序的。

### 15 消息（Message） ###

消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。

### 16 标签（Tag） ###

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。

## 特性(features) ##

--------------------

### 1 订阅与发布 ###

消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息，进而从该topic消费数据。

### 2 消息顺序 ###

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

顺序消息分为全局顺序消息与分区顺序消息，全局顺序是指某个Topic下的所有消息都要保证顺序；部分顺序消息只要保证每一组消息被顺序消费即可。

*  全局顺序  
    对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。  
    适用场景：性能要求不高，所有的消息严格按照 FIFO 原则进行消息发布和消费的场景
 *  分区顺序  
    对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。 Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念。  
    适用场景：性能要求高，以 sharding key 作为分区字段，在同一个区块中严格的按照 FIFO 原则进行消息发布和消费的场景。

### 3 消息过滤 ###

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

### 4 消息可靠性 ###

RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：

1.  Broker非正常关闭
2.  Broker异常Crash
3.  OS Crash
4.  机器掉电，但是能立即恢复供电情况
5.  机器无法开机（可能是cpu、主板、内存等关键设备损坏）
6.  磁盘设备损坏

1)、2)、3)、4) 四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。

5)、6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。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。注意，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消费消息失败通常可以认为有以下几种情况：

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

RocketMQ会为每个消费组都设置一个Topic名称为“%RETRY%+consumerGroup”的重试队列（这里需要注意的是，这个Topic的重试队列是针对消费组，而不是针对每个Topic设置的），用于暂时保存因为各种异常而导致Consumer端无法消费的消息。考虑到异常恢复起来需要一些时间，会为重试队列设置多个重试级别，每个重试级别都有与之对应的重新投递延时，重试次数越多投递延时就越大。RocketMQ对于重试消息的处理是先保存至Topic名称为“SCHEDULE\_TOPIC\_XXXX”的延迟队列中，后台定时任务按照对应的时间进行Delay后重新保存至“%RETRY%+consumerGroup”的重试队列中。

### 10 消息重投 ###

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

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

### 11 流量控制 ###

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

生产者流控：

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

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

消费者流控：

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

消费者流控的结果是降低拉取频率。

### 12 死信队列 ###

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

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

## 架构设计 ##

--------------------

### 1 技术架构 ###

![architecture][]

RocketMQ架构上主要分为四部分，如上图所示:

*  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的路由的信息。
 *  BrokerServer：Broker主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker包含了以下几个重要子模块。

1.  Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
2.  Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
3.  Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
4.  HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
5.  Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

![architecture][architecture 1]

### 2 部署架构 ###

![architecture][architecture 2]

#### RocketMQ 网络部署特点 ####

*  NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
 *  Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。
 *  Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
 *  Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从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建立连接通道，开始消费消息。

## 设计(design) ##

--------------------

### 1 消息存储 ###

![design][]

消息存储是RocketMQ中最为复杂和最为重要的一部分，本节将分别从RocketMQ的消息存储整体架构、PageCache与Mmap内存映射以及RocketMQ中两种不同的刷盘方式三方面来分别展开叙述。

#### 1.1 消息存储整体架构 ####

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；

(2) ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/\{topic\}/\{queueId\}/\{fileName\}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；

(3) IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME \\store\\index$\{fileName\}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。正因为如此，Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待30s的时间，只要这段时间内有新消息到达，将直接返回给消费端。这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

#### 1.2 页缓存与内存映射 ####

页缓存（PageCache)是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Deadline”（此时块存储采用SSD的话），随机读的性能也会有所提升。

另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存）。

#### 1.3 消息刷盘 ####

![design][design 1]

(1) 同步刷盘：如上图所示，只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。

(2) 异步刷盘：能够充分利用OS的PageCache的优势，只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。

### 2 通信机制 ###

RocketMQ消息队列集群主要包括NameServer、Broker(Master/Slave)、Producer、Consumer4个角色，基本通讯流程如下：

(1) Broker启动后需要完成一次将自己注册至NameServer的操作；随后每隔30s时间定时向NameServer上报Topic路由信息。

(2) 消息生产者Producer作为客户端发送消息时候，需要根据消息的Topic从本地缓存的TopicPublishInfoTable获取路由信息。如果没有则更新路由信息会从NameServer上重新拉取，同时Producer会默认每隔30s向NameServer拉取一次路由信息。

(3) 消息生产者Producer根据2）中获取的路由信息选择一个队列（MessageQueue）进行消息发送；Broker作为消息的接收者接收消息并落盘存储。

(4) 消息消费者Consumer根据2）中获取的路由信息，并再完成客户端的负载均衡后，选择其中的某一个或者某几个消息队列来拉取消息并进行消费。

从上面1）~3）中可以看出在消息生产者, Broker和NameServer之间都会发生通信（这里只说了MQ的部分通信），因此如何设计一个良好的网络通信模块在MQ中至关重要，它将决定RocketMQ集群整体的消息传输能力与最终的性能。

rocketmq-remoting 模块是 RocketMQ消息队列中负责网络通信的模块，它几乎被其他所有需要网络通信的模块（诸如rocketmq-client、rocketmq-broker、rocketmq-namesrv）所依赖和引用。为了实现客户端与服务器之间高效的数据请求与接收，RocketMQ消息队列自定义了通信协议并在Netty的基础之上扩展了通信模块。

#### 2.1 Remoting通信类结构 ####

![design][design 2]

#### 2.2 协议设计与编解码 ####

在Client和Server之间完成一次消息发送时，需要对发送的消息进行一个协议约定，因此就有必要自定义RocketMQ的消息协议。同时，为了高效地在网络中传输消息和对收到的消息读取，就需要对消息进行编解码。在RocketMQ中，RemotingCommand这个类在消息传输过程中对所有数据内容的封装，不但包含了所有的数据结构，还包含了编码解码操作。

<table> 
 <thead> 
  <tr> 
   <th>Header字段</th> 
   <th>类型</th> 
   <th>Request说明</th> 
   <th>Response说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>code</td> 
   <td>int</td> 
   <td>请求操作码，应答方根据不同的请求码进行不同的业务处理</td> 
   <td>应答响应码。0表示成功，非0则表示各种错误</td> 
  </tr> 
  <tr> 
   <td>language</td> 
   <td>LanguageCode</td> 
   <td>请求方实现的语言</td> 
   <td>应答方实现的语言</td> 
  </tr> 
  <tr> 
   <td>version</td> 
   <td>int</td> 
   <td>请求方程序的版本</td> 
   <td>应答方程序的版本</td> 
  </tr> 
  <tr> 
   <td>opaque</td> 
   <td>int</td> 
   <td>相当于requestId，在同一个连接上的不同请求标识码，与响应消息中的相对应</td> 
   <td>应答不做修改直接返回</td> 
  </tr> 
  <tr> 
   <td>flag</td> 
   <td>int</td> 
   <td>区分是普通RPC还是onewayRPC得标志</td> 
   <td>区分是普通RPC还是onewayRPC得标志</td> 
  </tr> 
  <tr> 
   <td>remark</td> 
   <td>String</td> 
   <td>传输自定义文本信息</td> 
   <td>传输自定义文本信息</td> 
  </tr> 
  <tr> 
   <td>extFields</td> 
   <td>HashMap&lt;String, String&gt;</td> 
   <td>请求自定义扩展信息</td> 
   <td>响应自定义扩展信息</td> 
  </tr> 
 </tbody> 
</table>

![design][design 3]

可见传输内容主要可以分为以下4部分：

(1) 消息长度：总长度，四个字节存储，占用一个int类型；

(2) 序列化类型&消息头长度：同样占用一个int类型，第一个字节表示序列化类型，后面三个字节表示消息头长度；

(3) 消息头数据：经过序列化后的消息头数据；

(4) 消息主体数据：消息主体的二进制字节数据内容；

#### 2.3 消息的通信方式和流程 ####

在RocketMQ消息队列中支持通信的方式主要有同步(sync)、异步(async)、单向(oneway)  
三种。其中“单向”通信模式相对简单，一般用在发送心跳包场景下，无需关注其Response。这里，主要介绍RocketMQ的异步通信流程。

![design][design 4]

#### 2.4 Reactor多线程设计 ####

RocketMQ的RPC通信采用Netty组件作为底层通信库，同样也遵循了Reactor多线程模型，同时又在这之上做了一些扩展和优化。

![design][design 5]

上面的框图中可以大致了解RocketMQ中NettyRemotingServer的Reactor 多线程模型。一个 Reactor 主线程（eventLoopGroupBoss，即为上面的1）负责监听 TCP网络连接请求，建立好连接，创建SocketChannel，并注册到selector上。RocketMQ的源码中会自动根据OS的类型选择NIO和Epoll，也可以通过参数配置）,然后监听真正的网络数据。拿到网络数据后，再丢给Worker线程池（eventLoopGroupSelector，即为上面的“N”，源码中默认设置为3），在真正执行业务逻辑之前需要进行SSL验证、编解码、空闲检查、网络连接管理，这些工作交给defaultEventExecutorGroup（即为上面的“M1”，源码中默认设置为8）去做。而处理业务操作放在业务线程池中执行，根据 RomotingCommand 的业务请求码code去processorTable这个本地缓存变量中找到对应的 processor，然后封装成task任务后，提交给对应的业务processor处理线程池来执行（sendMessageExecutor，以发送消息为例，即为上面的 “M2”）。从入口到业务逻辑的几个步骤中线程池一直再增加，这跟每一步逻辑复杂性相关，越复杂，需要的并发通道越宽。

<table> 
 <thead> 
  <tr> 
   <th>线程数</th> 
   <th>线程名</th> 
   <th>线程具体说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>NettyBoss_%d</td> 
   <td>Reactor 主线程</td> 
  </tr> 
  <tr> 
   <td>N</td> 
   <td>NettyServerEPOLLSelector_%d_%d</td> 
   <td>Reactor 线程池</td> 
  </tr> 
  <tr> 
   <td>M1</td> 
   <td>NettyServerCodecThread_%d</td> 
   <td>Worker线程池</td> 
  </tr> 
  <tr> 
   <td>M2</td> 
   <td>RemotingExecutorThread_%d</td> 
   <td>业务processor处理线程池</td> 
  </tr> 
 </tbody> 
</table>

### 3 消息过滤 ###

RocketMQ分布式消息队列的消息过滤方式有别于其它MQ中间件，是在Consumer端订阅消息时再做消息过滤的。RocketMQ这么做是在于其Producer端写入消息和Consumer端订阅消息采用分离存储的机制来实现的，Consumer端订阅消息是需要通过ConsumeQueue这个消息消费的逻辑队列拿到一个索引，然后再从CommitLog里面读取真正的消息实体内容，所以说到底也是还绕不开其存储结构。其ConsumeQueue的存储结构如下，可以看到其中有8个字节存储的Message Tag的哈希值，基于Tag的消息过滤正式基于这个字段值的。

![design][design 6]

主要支持如下2种的过滤方式  
(1) Tag过滤方式：Consumer端在订阅消息时除了指定Topic还可以指定TAG，如果一个消息有多个TAG，可以用||分隔。其中，Consumer端会将这个订阅请求构建成一个 SubscriptionData，发送一个Pull消息的请求给Broker端。Broker端从RocketMQ的文件存储层—Store读取数据之前，会用这些数据先构建一个MessageFilter，然后传给Store。Store从 ConsumeQueue读取到一条记录后，会用它记录的消息tag hash值去做过滤，由于在服务端只是根据hashcode进行判断，无法精确对tag原始字符串进行过滤，故在消息消费端拉取到消息后，还需要对消息的原始tag字符串进行比对，如果不同，则丢弃该消息，不进行消息消费。

(2) SQL92的过滤方式：这种方式的大致做法和上面的Tag过滤方式一样，只是在Store层的具体过滤过程不太一样，真正的 SQL expression 的构建和执行由rocketmq-filter模块负责的。每次过滤都去执行SQL表达式会影响效率，所以RocketMQ使用了BloomFilter避免了每次都去执行。SQL92的表达式上下文为消息的属性。

### 4 负载均衡 ###

RocketMQ中的负载均衡都在Client端完成，具体来说的话，主要可以分为Producer端发送消息时候的负载均衡和Consumer端订阅消息的负载均衡。

#### 4.1 Producer的负载均衡 ####

Producer端在发送消息的时候，会先根据Topic找到指定的TopicPublishInfo，在获取了TopicPublishInfo路由信息后，RocketMQ的客户端在默认方式下selectOneMessageQueue()方法会从TopicPublishInfo中的messageQueueList中选择一个队列（MessageQueue）进行发送消息。具体的容错策略均在MQFaultStrategy这个类中定义。这里有一个sendLatencyFaultEnable开关变量，如果开启，在随机递增取模的基础上，再过滤掉not available的Broker代理。所谓的"latencyFaultTolerance"，是指对之前失败的，按一定的时间做退避。例如，如果上次请求的latency超过550Lms，就退避3000Lms；超过1000L，就退避60000L；如果关闭，采用随机递增取模的方式选择一个队列（MessageQueue）来发送消息，latencyFaultTolerance机制是实现消息发送高可用的核心关键所在。

#### 4.2 Consumer的负载均衡 ####

在RocketMQ中，Consumer端的两种消费模式（Push/Pull）都是基于拉模式来获取消息的，而在Push模式只是对pull模式的一种封装，其本质实现为消息拉取线程在从服务器拉取到一批消息后，然后提交到消息消费线程池后，又“马不停蹄”的继续向服务器再次尝试拉取消息。如果未拉取到消息，则延迟一下又继续拉取。在两种基于拉模式的消费方式（Push/Pull）中，均需要Consumer端在知道从Broker端的哪一个消息队列—队列中去获取消息。因此，有必要在Consumer端来做负载均衡，即Broker端中多个MessageQueue分配给同一个ConsumerGroup中的哪些Consumer消费。

1、Consumer端的心跳包发送

在Consumer启动后，它就会通过定时任务不断地向RocketMQ集群中的所有Broker实例发送心跳包（其中包含了，消息消费分组名称、订阅关系集合、消息通信模式和客户端id的值等信息）。Broker端在收到Consumer的心跳消息后，会将它维护在ConsumerManager的本地缓存变量—consumerTable，同时并将封装后的客户端网络通道信息保存在本地缓存变量—channelInfoTable中，为之后做Consumer端的负载均衡提供可以依据的元数据信息。

2、Consumer端实现负载均衡的核心类—RebalanceImpl

在Consumer实例的启动流程中的启动MQClientInstance实例部分，会完成负载均衡服务线程—RebalanceService的启动（每隔20s执行一次）。通过查看源码可以发现，RebalanceService线程的run()方法最终调用的是RebalanceImpl类的rebalanceByTopic()方法，该方法是实现Consumer端负载均衡的核心。这里，rebalanceByTopic()方法会根据消费者通信类型为“广播模式”还是“集群模式”做不同的逻辑处理。这里主要来看下集群模式下的主要处理流程：

(1) 从rebalanceImpl实例的本地缓存变量—topicSubscribeInfoTable中，获取该Topic主题下的消息消费队列集合（mqSet）；

(2) 根据topic和consumerGroup为参数调用mQClientFactory.findConsumerIdList()方法向Broker端发送获取该消费组下消费者Id列表的RPC通信请求（Broker端基于前面Consumer端上报的心跳包数据而构建的consumerTable做出响应返回，业务请求码：GET\_CONSUMER\_LIST\_BY\_GROUP）；

(3) 先对Topic下的消息消费队列、消费者Id排序，然后用消息队列分配策略算法（默认为：消息队列的平均分配算法），计算出待拉取的消息队列。这里的平均分配算法，类似于分页的算法，将所有MessageQueue排好序类似于记录，将所有消费端Consumer排好序类似页数，并求出每一页需要包含的平均size和每个页面记录的范围range，最后遍历整个range而计算出当前Consumer端应该分配到的记录（这里即为：MessageQueue）。

![design][design 7]

(4) 然后，调用updateProcessQueueTableInRebalance()方法，具体的做法是，先将分配到的消息队列集合（mqSet）与processQueueTable做一个过滤比对。

![design][design 8]

*  上图中processQueueTable标注的红色部分，表示与分配到的消息队列集合mqSet互不包含。将这些队列设置Dropped属性为true，然后查看这些队列是否可以移除出processQueueTable缓存变量，这里具体执行removeUnnecessaryMessageQueue()方法，即每隔1s 查看是否可以获取当前消费处理队列的锁，拿到的话返回true。如果等待1s后，仍然拿不到当前消费处理队列的锁则返回false。如果返回true，则从processQueueTable缓存变量中移除对应的Entry；
 *  上图中processQueueTable的绿色部分，表示与分配到的消息队列集合mqSet的交集。判断该ProcessQueue是否已经过期了，在Pull模式的不用管，如果是Push模式的，设置Dropped属性为true，并且调用removeUnnecessaryMessageQueue()方法，像上面一样尝试移除Entry；

最后，为过滤后的消息队列集合（mqSet）中的每个MessageQueue创建一个ProcessQueue对象并存入RebalanceImpl的processQueueTable队列中（其中调用RebalanceImpl实例的computePullFromWhere(MessageQueue mq)方法获取该MessageQueue对象的下一个进度消费值offset，随后填充至接下来要创建的pullRequest对象属性中），并创建拉取请求对象—pullRequest添加到拉取列表—pullRequestList中，最后执行dispatchPullRequest()方法，将Pull消息的请求对象PullRequest依次放入PullMessageService服务线程的阻塞队列pullRequestQueue中，待该服务线程取出后向Broker端发起Pull消息的请求。其中，可以重点对比下，RebalancePushImpl和RebalancePullImpl两个实现类的dispatchPullRequest()方法不同，RebalancePullImpl类里面的该方法为空，这样子也就回答了上一篇中最后的那道思考题了。

消息消费队列在同一消费组不同消费者之间的负载均衡，其核心设计理念是在一个消息消费队列在同一时间只允许被同一消费组内的一个消费者消费，一个消息消费者能同时消费多个消息队列。

### 5 事务消息 ###

Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

![design][design 9]

#### 5.1 RocketMQ事务消息流程概要 ####

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。

1.事务消息发送及提交：

(1) 发送消息（half消息）。

(2) 服务端响应消息写入结果。

(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。

(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

2.补偿流程：

(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”

(2) Producer收到回查消息，检查回查消息对应的本地事务的状态

(3) 根据本地事务状态，重新Commit或者Rollback

其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

#### 5.2 RocketMQ事务消息设计 ####

1.事务消息在一阶段对用户不可见

在RocketMQ事务消息的主要流程中，一阶段的消息如何对用户不可见。其中，事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的。那么，如何做到写入消息但是对用户不可见呢？RocketMQ事务消息的做法是：如果消息是half消息，将备份原消息的主题与消息消费队列，然后改变主题为RMQ\_SYS\_TRANS\_HALF\_TOPIC。由于消费组未订阅该主题，故消费端无法消费half类型的消息，然后RocketMQ会开启一个定时任务，从Topic为RMQ\_SYS\_TRANS\_HALF\_TOPIC中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。

在RocketMQ中，消息在服务端的存储结构如下，每条消息都会有对应的索引信息，Consumer通过ConsumeQueue这个二级索引来读取消息实体内容，其流程如下：

![design][design 10]

RocketMQ的具体实现策略是：写入的如果事务消息，对消息的Topic和Queue等属性进行替换，同时将原来的Topic和Queue信息存储到消息的属性中，正因为消息主题被替换，故消息并不会转发到该原主题的消息消费队列，消费者无法感知消息的存在，不会消费。其实改变消息主题是RocketMQ的常用“套路”，回想一下延时消息的实现机制。

2.Commit和Rollback操作以及Op消息的引入

在完成一阶段写入一条对用户不可见的消息后，二阶段如果是Commit操作，则需要让消息对用户可见；如果是Rollback则需要撤销一阶段的消息。先说Rollback的情况。对于Rollback，本身一阶段的消息对用户是不可见的，其实不需要真正撤销消息（实际上RocketMQ也无法去真正的删除一条消息，因为是顺序写文件的）。但是区别于这条消息没有确定状态（Pending状态，事务悬而未决），需要一个操作来标识这条消息的最终状态。RocketMQ事务消息方案中引入了Op消息的概念，用Op消息标识事务消息已经确定的状态（Commit或者Rollback）。如果一条事务消息没有对应的Op消息，说明这个事务的状态还无法确定（可能是二阶段失败了）。引入Op消息后，事务消息无论是Commit或者Rollback都会记录一个Op操作。Commit相对于Rollback只是在写入Op消息前创建Half消息的索引。

3.Op消息的存储和对应关系

RocketMQ将Op消息写入到全局一个特定的Topic中通过源码中的方法—TransactionalMessageUtil.buildOpTopic()；这个Topic是一个内部的Topic（像Half消息的Topic一样），不会被用户消费。Op消息的内容为对应的Half消息的存储的Offset，这样通过Op消息能索引到Half消息进行后续的回查操作。

![design][design 11]

4.Half消息的索引构建

在执行二阶段Commit操作时，需要构建出Half消息的索引。一阶段的Half消息由于是写到一个特殊的Topic，所以二阶段构建索引时需要读取出Half消息，并将Topic和Queue替换成真正的目标的Topic和Queue，之后通过一次普通消息的写入操作来生成一条对用户可见的消息。所以RocketMQ事务消息二阶段其实是利用了一阶段存储的消息的内容，在二阶段时恢复出一条完整的普通消息，然后走一遍消息写入流程。

5.如何处理二阶段失败的消息？

如果在RocketMQ事务消息的二阶段过程中失败了，例如在做Commit操作时，出现网络问题导致Commit失败，那么需要通过一定的策略使这条消息最终被Commit。RocketMQ采用了一种补偿机制，称为“回查”。Broker端对未确定状态的消息发起回查，将消息发送到对应的Producer端（同一个Group的Producer），由Producer根据消息来检查本地事务的状态，进而执行Commit或者Rollback。Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint（记录那些事务消息的状态是确定的）。

值得注意的是，rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，rocketmq默认回滚该消息。

### 6 消息查询 ###

RocketMQ支持按照下面两种维度（“按照Message Id查询消息”、“按照Message Key查询消息”）进行消息查询。

#### 6.1 按照MessageId查询消息 ####

RocketMQ中的MessageId的长度总共有16字节，其中包含了消息存储主机地址（IP地址和端口），消息Commit Log offset。“按照MessageId查询消息”在RocketMQ中具体做法是：Client端从MessageId中解析出Broker的地址（IP地址和端口）和Commit Log的偏移地址后封装成一个RPC请求后通过Remoting通信层发送（业务请求码：VIEW\_MESSAGE\_BY\_ID）。Broker端走的是QueryMessageProcessor，读取消息的过程用其中的 commitLog offset 和 size 去 commitLog 中找到真正的记录并解析成一个完整的消息返回。

#### 6.2 按照Message Key查询消息 ####

“按照Message Key查询消息”，主要是基于RocketMQ的IndexFile索引文件来实现的。RocketMQ的索引文件逻辑结构，类似JDK中HashMap的实现。索引文件的具体结构如下：

![design][design 12]

IndexFile索引文件为用户提供通过“按照Message Key查询消息”的消息索引查询服务，IndexFile文件的存储位置是：$HOME\\store\\index$\{fileName\}，文件名fileName是以创建时的时间戳命名的，文件大小是固定的，等于40+500W\*4+2000W\*20= 420000040个字节大小。如果消息的properties中设置了UNIQ\_KEY这个属性，就用 topic + “\#” + UNIQ\_KEY的value作为 key 来做写入操作。如果消息设置了KEYS属性（多个KEY以空格分隔），也会用 topic + “\#” + KEY 来做索引。

其中的索引数据包含了Key Hash/CommitLog Offset/Timestamp/NextIndex offset 这四个字段，一共20 Byte。NextIndex offset 即前面读出来的 slotValue，如果有 hash冲突，就可以用这个字段将所有冲突的索引用链表的方式串起来了。Timestamp记录的是消息storeTimestamp之间的差，并不是一个绝对的时间。整个Index File的结构如图，40 Byte 的Header用于保存一些总的统计信息，4\*500W的 Slot Table并不保存真正的索引数据，而是保存每个槽位对应的单向链表的头。20\*2000W 是真正的索引数据，即一个 Index File 可以保存 2000W个索引。

“按照Message Key查询消息”的方式，RocketMQ的具体做法是，主要通过Broker端的QueryMessageProcessor业务处理器来查询，读取消息的过程就是用topic和key找到IndexFile索引文件中的一条记录，根据其中的commitLog offset从CommitLog文件中读取消息的实体内容。

## 样例 ##

--------------------

### 1 基本样例 ###

在基本样例中我们提供如下的功能场景：

*  使用RocketMQ发送三种类型的消息：同步消息、异步消息和单向消息。其中前两种消息是可靠的，因为会有发送是否成功的应答。
 *  使用RocketMQ来消费接收到的消息。

#### 1.1 加入依赖： ####

`maven:`

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.3.0</version>
    </dependency>

`gradle`

compile 'org.apache.rocketmq:rocketmq-client:4.3.0'

#### 1.2 消息发送 ####

##### 1、Producer端发送同步消息 #####

这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

public class SyncProducer { 
    	public static void main(String[] args) throws Exception { 
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
        	producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
        	for (int i = 0; i < 100; i++) { 
        	    // 创建消息，并指定Topic，Tag和消息体
        	    Message msg = new Message("TopicTest" /* Topic */,
            	"TagA" /* Tag */,
            	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            	);
            	// 发送消息到一个Broker
                SendResult sendResult = producer.send(msg);
                // 通过sendResult返回消息是否成功送达
                System.out.printf("%s%n", sendResult);
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }

##### 2、发送异步消息 #####

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。

public class AsyncProducer { 
    	public static void main(String[] args) throws Exception { 
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
            producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);
    	
    	int messageCount = 100;
            // 根据消息数量实例化倒计时计算器
    	final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
        	for (int i = 0; i < messageCount; i++) { 
                    final int index = i;
                	// 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // SendCallback接收异步返回结果的回调
                    producer.send(msg, new SendCallback() { 
                        @Override
                        public void onSuccess(SendResult sendResult) { 
                            System.out.printf("%-10d OK %s %n", index,
                                sendResult.getMsgId());
                        }
                        @Override
                        public void onException(Throwable e) { 
          	              System.out.printf("%-10d Exception %s %n", index, e);
          	              e.printStackTrace();
                        }
                	});
        	}
    	// 等待5s
    	countDownLatch.await(5, TimeUnit.SECONDS);
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }

##### 3、单向发送消息 #####

这种方式主要用在不特别关心发送结果的场景，例如日志发送。

public class OnewayProducer { 
    	public static void main(String[] args) throws Exception{ 
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        	// 设置NameServer的地址
            producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
        	for (int i = 0; i < 100; i++) { 
            	// 创建消息，并指定Topic，Tag和消息体
            	Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            	);
            	// 发送单向消息，没有任何返回结果
            	producer.sendOneway(msg);
    
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }

#### 1.3 消费消息 ####

public class Consumer { 
    
    	public static void main(String[] args) throws InterruptedException, MQClientException { 
    
        	// 实例化消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    
        	// 设置NameServer的地址
            consumer.setNamesrvAddr("localhost:9876");
    
        	// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
            consumer.subscribe("TopicTest", "*");
        	// 注册回调实现类来处理从broker拉取回来的消息
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    // 标记该消息已经被成功消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者实例
            consumer.start();
            System.out.printf("Consumer Started.%n");
    	}
    }

### 2 顺序消息样例 ###

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。

顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。

#### 2.1 顺序消息生产 ####

package org.apache.rocketmq.example.order2;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageQueue;
    
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    
    /** * Producer，发送顺序消息 */
    public class Producer { 
    
       public static void main(String[] args) throws Exception { 
           DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    
           producer.setNamesrvAddr("127.0.0.1:9876");
    
           producer.start();
    
           String[] tags = new String[]{ "TagA", "TagC", "TagD"};
    
           // 订单列表
           List<OrderStep> orderList = new Producer().buildOrders();
    
           Date date = new Date();
           SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
           String dateStr = sdf.format(date);
           for (int i = 0; i < 10; i++) { 
               // 加个时间前缀
               String body = dateStr + " Hello RocketMQ " + orderList.get(i);
               Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
    
               SendResult sendResult = producer.send(msg, new MessageQueueSelector() { 
                   @Override
                   public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { 
                       Long id = (Long) arg;  //根据订单id选择发送queue
                       long index = id % mqs.size();
                       return mqs.get((int) index);
                   }
               }, orderList.get(i).getOrderId());//订单id
    
               System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
                   sendResult.getSendStatus(),
                   sendResult.getMessageQueue().getQueueId(),
                   body));
           }
    
           producer.shutdown();
       }
    
       /** * 订单的步骤 */
       private static class OrderStep { 
           private long orderId;
           private String desc;
    
           public long getOrderId() { 
               return orderId;
           }
    
           public void setOrderId(long orderId) { 
               this.orderId = orderId;
           }
    
           public String getDesc() { 
               return desc;
           }
    
           public void setDesc(String desc) { 
               this.desc = desc;
           }
    
           @Override
           public String toString() { 
               return "OrderStep{" +
                   "orderId=" + orderId +
                   ", desc='" + desc + '\'' +
                   '}';
           }
       }
    
       /** * 生成模拟订单数据 */
       private List<OrderStep> buildOrders() { 
           List<OrderStep> orderList = new ArrayList<OrderStep>();
    
           OrderStep orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111039L);
           orderDemo.setDesc("创建");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111065L);
           orderDemo.setDesc("创建");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111039L);
           orderDemo.setDesc("付款");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103117235L);
           orderDemo.setDesc("创建");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111065L);
           orderDemo.setDesc("付款");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103117235L);
           orderDemo.setDesc("付款");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111065L);
           orderDemo.setDesc("完成");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111039L);
           orderDemo.setDesc("推送");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103117235L);
           orderDemo.setDesc("完成");
           orderList.add(orderDemo);
    
           orderDemo = new OrderStep();
           orderDemo.setOrderId(15103111039L);
           orderDemo.setDesc("完成");
           orderList.add(orderDemo);
    
           return orderList;
       }
    }

#### 2.2 顺序消费消息 ####

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import java.util.List;
    
    package org.apache.rocketmq.example.order2;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /** * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交） */
    public class ConsumerInOrder { 
    
       public static void main(String[] args) throws Exception { 
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
           consumer.setNamesrvAddr("127.0.0.1:9876");
           /** * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> * 如果非第一次启动，那么按照上次消费的位置继续消费 */
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
           consumer.subscribe("TopicTest", "TagA || TagC || TagD");
    
           consumer.registerMessageListener(new MessageListenerOrderly() { 
    
               Random random = new Random();
    
               @Override
               public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { 
                   context.setAutoCommit(true);
                   for (MessageExt msg : msgs) { 
                       // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                       System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
                   }
    
                   try { 
                       //模拟业务逻辑处理中...
                       TimeUnit.SECONDS.sleep(random.nextInt(10));
                   } catch (Exception e) { 
                       e.printStackTrace();
                   }
                   return ConsumeOrderlyStatus.SUCCESS;
               }
           });
    
           consumer.start();
    
           System.out.println("Consumer Started.");
       }
    }

### 3 延时消息样例 ###

#### 3.1 启动消费者等待传入订阅消息 ####

#### 3.2 发送延时消息 ####

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.message.Message;
    
    public class ScheduledMessageProducer { 
       public static void main(String[] args) throws Exception { 
          // 实例化一个生产者来产生延时消息
          DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
          // 启动生产者
          producer.start();
          int totalMessagesToSend = 100;
          for (int i = 0; i < totalMessagesToSend; i++) { 
              Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
              // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
              message.setDelayTimeLevel(3);
              // 发送消息
              producer.send(message);
          }
           // 关闭生产者
          producer.shutdown();
      }
    }

#### 3.3 验证 ####

您将会看到消息的消费比存储时间晚10秒。

#### 3.4 延时消息的使用场景 ####

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

#### 3.5 延时消息的使用限制 ####

// org/apache/rocketmq/store/config/MessageStoreConfig.java
    
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18  
消息消费失败会进入延时消息队列，消息发送时间与设置的延时等级和重试次数有关，详见代码`SendMessageProcessor.java`

### 4 批量消息样例 ###

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。

#### 4.1 发送批量消息 ####

如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：

String topic = "BatchTest";
    List<Message> messages = new ArrayList<>();
    messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
    messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
    messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
    try { 
       producer.send(messages);
    } catch (Exception e) { 
       e.printStackTrace();
       //处理error
    }

#### 4.2 消息列表分割 ####

复杂度只有当你发送大批量时才会增长，你可能不确定它是否超过了大小限制（4MB）。这时候你最好把你的消息列表分割一下：

public class ListSplitter implements Iterator<List<Message>> { 
       private final int SIZE_LIMIT = 1024 * 1024 * 4;
       private final List<Message> messages;
       private int currIndex;
       public ListSplitter(List<Message> messages) { 
               this.messages = messages;
       }
       @Override public boolean hasNext() { 
           return currIndex < messages.size();
       }
       @Override public List<Message> next() { 
           int nextIndex = currIndex;
           int totalSize = 0;
           for (; nextIndex < messages.size(); nextIndex++) { 
               Message message = messages.get(nextIndex);
               int tmpSize = message.getTopic().length() + message.getBody().length;
               Map<String, String> properties = message.getProperties();
               for (Map.Entry<String, String> entry : properties.entrySet()) { 
                   tmpSize += entry.getKey().length() + entry.getValue().length();
               }
               tmpSize = tmpSize + 20; // 增加日志的开销20字节
               if (tmpSize > SIZE_LIMIT) { 
                   //单个消息超过了最大的限制
                   //忽略,否则会阻塞分裂的进程
                   if (nextIndex - currIndex == 0) { 
                      //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
                      nextIndex++;
                   }
                   break;
               }
               if (tmpSize + totalSize > SIZE_LIMIT) { 
                   break;
               } else { 
                   totalSize += tmpSize;
               }
    
           }
           List<Message> subList = messages.subList(currIndex, nextIndex);
           currIndex = nextIndex;
           return subList;
       }
    }
    //把大的消息分裂成若干个小的消息
    ListSplitter splitter = new ListSplitter(messages);
    while (splitter.hasNext()) { 
      try { 
          List<Message>  listItem = splitter.next();
          producer.send(listItem);
      } catch (Exception e) { 
          e.printStackTrace();
          //处理error
      }
    }

### 5 过滤消息样例 ###

在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息。例如：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
    consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：

------------
    | message  |
    |----------|  a > 5 AND b = 'abc'
    | a = 10   |  --------------------> Gotten
    | b = 'abc'|
    | c = true |
    ------------
    ------------
    | message  |
    |----------|   a > 5 AND b = 'abc'
    | a = 1    |  --------------------> Missed
    | b = 'abc'|
    | c = true |
    ------------

#### 5.1 基本语法 ####

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

*  数值比较，比如：**>，>=，<，<=，BETWEEN，=；**
 *  字符比较，比如：**=，<>，IN；**
 *  **IS NULL** 或者 **IS NOT NULL；**
 *  逻辑符号 **AND，OR，NOT；**

常量支持类型为：

*  数值，比如：**123，3.1415；**
 *  字符，比如：**‘abc’，必须用单引号包裹起来；**
 *  **NULL**，特殊的常量
 *  布尔值，**TRUE** 或 **FALSE**

只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：

public void subscribe(finalString topic, final MessageSelector messageSelector)

#### 5.2 使用样例 ####

##### 1、生产者样例 #####

发送消息时，你能通过`putUserProperty`来设置消息的属性

DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    producer.start();
    Message msg = new Message("TopicTest",
       tag,
       ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
    );
    // 设置一些属性
    msg.putUserProperty("a", String.valueOf(i));
    SendResult sendResult = producer.send(msg);
    
    producer.shutdown();

##### 2、消费者样例 #####

用MessageSelector.bySql来使用sql筛选消息

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
    // 只有订阅的消息有这个属性a, a >=0 and a <= 3
    consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
    consumer.registerMessageListener(new MessageListenerConcurrently() { 
       @Override
       public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }
    });
    consumer.start();

### 6 消息事务样例 ###

事务消息共有三种状态，提交状态、回滚状态、中间状态：

*  TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
 *  TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
 *  TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

#### 6.1 发送事务消息样例 ####

##### 1、创建事务性生产者 #####

使用 `TransactionMQProducer`类创建生产者，并指定唯一的 `ProducerGroup`，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import java.util.List;
    public class TransactionProducer { 
       public static void main(String[] args) throws MQClientException, InterruptedException { 
           TransactionListener transactionListener = new TransactionListenerImpl();
           TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
           ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() { 
               @Override
               public Thread newThread(Runnable r) { 
                   Thread thread = new Thread(r);
                   thread.setName("client-transaction-msg-check-thread");
                   return thread;
               }
           });
           producer.setExecutorService(executorService);
           producer.setTransactionListener(transactionListener);
           producer.start();
           String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE"};
           for (int i = 0; i < 10; i++) { 
               try { 
                   Message msg =
                       new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                           ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                   SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                   System.out.printf("%s%n", sendResult);
                   Thread.sleep(10);
               } catch (MQClientException | UnsupportedEncodingException e) { 
                   e.printStackTrace();
               }
           }
           for (int i = 0; i < 100000; i++) { 
               Thread.sleep(1000);
           }
           producer.shutdown();
       }
    }

##### 2、实现事务的监听接口 #####

当发送半消息成功时，我们使用 `executeLocalTransaction` 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。`checkLocalTranscation` 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。

public class TransactionListenerImpl implements TransactionListener { 
      private AtomicInteger transactionIndex = new AtomicInteger(0);
      private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
      @Override
      public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
          int value = transactionIndex.getAndIncrement();
          int status = value % 3;
          localTrans.put(msg.getTransactionId(), status);
          return LocalTransactionState.UNKNOW;
      }
      @Override
      public LocalTransactionState checkLocalTransaction(MessageExt msg) { 
          Integer status = localTrans.get(msg.getTransactionId());
          if (null != status) { 
              switch (status) { 
                  case 0:
                      return LocalTransactionState.UNKNOW;
                  case 1:
                      return LocalTransactionState.COMMIT_MESSAGE;
                  case 2:
                      return LocalTransactionState.ROLLBACK_MESSAGE;
              }
          }
          return LocalTransactionState.COMMIT_MESSAGE;
      }
    }

#### 6.2 事务消息使用上的限制 ####

1.  事务消息不支持延时消息和批量消息。
2.  为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 `transactionCheckMax`参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = `transactionCheckMax` ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 `AbstractTransactionCheckListener` 类来修改这个行为。
3.  事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK\_IMMUNITY\_TIME\_IN\_SECONDS 来改变这个限制，该参数优先于 `transactionMsgTimeout` 参数。
4.  事务性消息可能不止一次被检查或消费。
5.  提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
6.  事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

### 7 Logappender样例 ###

RocketMQ日志提供log4j、log4j2和logback日志框架作为业务应用，下面是配置样例

#### 7.1 log4j样例 ####

按下面样例使用log4j属性配置

log4j.appender.mq=org.apache.rocketmq.logappender.log4j.RocketmqLog4jAppender
    log4j.appender.mq.Tag=yourTag
    log4j.appender.mq.Topic=yourLogTopic
    log4j.appender.mq.ProducerGroup=yourLogGroup
    log4j.appender.mq.NameServerAddress=yourRocketmqNameserverAddress
    log4j.appender.mq.layout=org.apache.log4j.PatternLayout
    log4j.appender.mq.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-4r [%t] (%F:%L) %-5p - %m%n

按下面样例使用log4j xml配置来使用异步添加日志

#### 7.2 log4j2样例 ####

用log4j2时，配置如下，如果想要非阻塞，只需要使用异步添加引用即可

#### 7.3 logback样例 ####

<appender name="mqAppender1"class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">
      <tag>yourTag</tag>
      <topic>yourLogTopic</topic>
      <producerGroup>yourLogGroup</producerGroup>
      <nameServerAddress>yourRocketmqNameserverAddress</nameServerAddress>
      <layout>
          <pattern>%date %p %t - %m%n</pattern>
      </layout>
    </appender>
    <appender name="mqAsyncAppender1"class="ch.qos.logback.classic.AsyncAppender">
      <queueSize>1024</queueSize>
      <discardingThreshold>80</discardingThreshold>
      <maxFlushTime>2000</maxFlushTime>
      <neverBlock>true</neverBlock>
      <appender-ref ref="mqAppender1"/>
    </appender>

### 8 OpenMessaging样例 ###

[OpenMessaging][]旨在建立消息和流处理规范，以为金融、电子商务、物联网和大数据领域提供通用框架及工业级指导方案。在分布式异构环境中，设计原则是面向云、简单、灵活和独立于语言。符合这些规范将帮助企业方便的开发跨平台和操作系统的异构消息传递应用程序。提供了openmessaging-api 0.3.0-alpha的部分实现，下面的示例演示如何基于OpenMessaging访问RocketMQ。

#### 8.1 OMSProducer样例 ####

下面的示例演示如何在同步、异步或单向传输中向RocketMQ代理发送消息。

import io.openmessaging.Future;
    import io.openmessaging.FutureListener;
    import io.openmessaging.Message;
    import io.openmessaging.MessagingAccessPoint;
    import io.openmessaging.OMS;
    import io.openmessaging.producer.Producer;
    import io.openmessaging.producer.SendResult;
    import java.nio.charset.Charset;
    import java.util.concurrent.CountDownLatch;
    
    public class SimpleProducer { 
        public static void main(String[] args) { 
           final MessagingAccessPoint messagingAccessPoint =
               OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
           final Producer producer = messagingAccessPoint.createProducer();
           messagingAccessPoint.startup();
           System.out.printf("MessagingAccessPoint startup OK%n");
           producer.startup();
           System.out.printf("Producer startup OK%n");
           { 
               Message message = producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
               SendResult sendResult = producer.send(message);
               //final Void aVoid = result.get(3000L);
               System.out.printf("Send async message OK, msgId: %s%n", sendResult.messageId());
           }
           final CountDownLatch countDownLatch = new CountDownLatch(1);
           { 
               final Future<SendResult> result = producer.sendAsync(producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
               result.addListener(new FutureListener<SendResult>() { 
                   @Override
                   public void operationComplete(Future<SendResult> future) { 
                       if (future.getThrowable() != null) { 
                           System.out.printf("Send async message Failed, error: %s%n", future.getThrowable().getMessage());
                       } else { 
                           System.out.printf("Send async message OK, msgId: %s%n", future.get().messageId());
                       }
                       countDownLatch.countDown();
                   }
               });
           }
           { 
               producer.sendOneway(producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
               System.out.printf("Send oneway message OK%n");
           }
           try { 
               countDownLatch.await();
               Thread.sleep(500); // 等一些时间来发送消息
           } catch (InterruptedException ignore) { 
           }
           producer.shutdown();
       }
    }

#### 8.2 OMSPullConsumer ####

用OMS PullConsumer 来从指定的队列中拉取消息

import io.openmessaging.Message;
    import io.openmessaging.MessagingAccessPoint;
    import io.openmessaging.OMS;
    import io.openmessaging.OMSBuiltinKeys;
    import io.openmessaging.consumer.PullConsumer;
    import io.openmessaging.producer.Producer;
    import io.openmessaging.producer.SendResult;
    
    public class SimplePullConsumer { 
        public static void main(String[] args) { 
           final MessagingAccessPoint messagingAccessPoint =
               OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
           messagingAccessPoint.startup();
           final Producer producer = messagingAccessPoint.createProducer();
           final PullConsumer consumer = messagingAccessPoint.createPullConsumer(
               OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
           messagingAccessPoint.startup();
           System.out.printf("MessagingAccessPoint startup OK%n");
           final String queueName = "TopicTest";
           producer.startup();
           Message msg = producer.createBytesMessage(queueName, "Hello Open Messaging".getBytes());
           SendResult sendResult = producer.send(msg);
           System.out.printf("Send Message OK. MsgId: %s%n", sendResult.messageId());
           producer.shutdown();
           consumer.attachQueue(queueName);
           consumer.startup();
           System.out.printf("Consumer startup OK%n");
           // 运行直到发现一个消息被发送了
           boolean stop = false;
           while (!stop) { 
               Message message = consumer.receive();
               if (message != null) { 
                   String msgId = message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID);
                   System.out.printf("Received one message: %s%n", msgId);
                   consumer.ack(msgId);
                   if (!stop) { 
                       stop = msgId.equalsIgnoreCase(sendResult.messageId());
                   }
               } else { 
                   System.out.printf("Return without any message%n");
               }
           }
           consumer.shutdown();
           messagingAccessPoint.shutdown();
       }
    }

#### 8.3 OMSPushConsumer ####

以下示范如何将 OMS PushConsumer 添加到指定的队列，并通过 MessageListener 消费这些消息。

import io.openmessaging.Message;
    import io.openmessaging.MessagingAccessPoint;
    import io.openmessaging.OMS;
    import io.openmessaging.OMSBuiltinKeys;
    import io.openmessaging.consumer.MessageListener;
    import io.openmessaging.consumer.PushConsumer;
    
    public class SimplePushConsumer { 
        public static void main(String[] args) { 
           final MessagingAccessPoint messagingAccessPoint = OMS
               .getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
           final PushConsumer consumer = messagingAccessPoint.
               createPushConsumer(OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
           messagingAccessPoint.startup();
           System.out.printf("MessagingAccessPoint startup OK%n");
           Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { 
               @Override
               public void run() { 
                   consumer.shutdown();
                   messagingAccessPoint.shutdown();
               }
           }));
           consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() { 
               @Override
               public void onReceived(Message message, Context context) { 
                   System.out.printf("Received one message: %s%n", message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID));
                   context.ack();
               }
           });
           consumer.startup();
           System.out.printf("Consumer startup OK%n");
       }
    }

## 最佳实践 ##

### 1 生产者 ###

#### 1.1 发送消息注意事项 ####

##### 1 Tags的使用 #####

一个应用尽可能用一个Topic，而消息子类型则可以用tags来标识。tags可以由应用自由设置，只有生产者在发送消息设置了tags，消费方在订阅消息时才可以利用tags通过broker做消息过滤：message.setTags(“TagA”)。

##### 2 Keys的使用 #####

每个消息在业务层面的唯一标识码要设置到keys字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过topic、key来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。

// 订单Id 
       String orderId = "20034568923546";   
       message.setKeys(orderId);

##### 3 日志的打印 #####

消息发送成功或者失败要打印消息日志，务必要打印SendResult和key字段。send消息方法只要不抛异常，就代表发送成功。发送成功会有多个状态，在sendResult里定义。以下对每个状态进行说明：

*  **SEND\_OK**

消息发送成功。要注意的是消息发送成功也不意味着它是可靠的。要确保不会丢失任何消息，还应启用同步Master服务器或同步刷盘，即SYNC\_MASTER或SYNC\_FLUSH。

*  **FLUSH\_DISK\_TIMEOUT**

消息发送成功但是服务器刷盘超时。此时消息已经进入服务器队列（内存），只有服务器宕机，消息才会丢失。消息存储配置参数中可以设置刷盘方式和同步刷盘时间长度，如果Broker服务器设置了刷盘方式为同步刷盘，即FlushDiskType=SYNC\_FLUSH（默认为异步刷盘方式），当Broker服务器未在同步刷盘时间内（默认为5s）完成刷盘，则将返回该状态——刷盘超时。

*  **FLUSH\_SLAVE\_TIMEOUT**

消息发送成功，但是服务器同步到Slave时超时。此时消息已经进入服务器队列，只有服务器宕机，消息才会丢失。如果Broker服务器的角色是同步Master，即SYNC\_MASTER（默认是异步Master即ASYNC\_MASTER），并且从Broker服务器未在同步刷盘时间（默认为5秒）内完成与主服务器的同步，则将返回该状态——数据同步到Slave服务器超时。

*  **SLAVE\_NOT\_AVAILABLE**

消息发送成功，但是此时Slave不可用。如果Broker服务器的角色是同步Master，即SYNC\_MASTER（默认是异步Master服务器即ASYNC\_MASTER），但没有配置slave Broker服务器，则将返回该状态——无Slave服务器可用。

#### 1.2 消息发送失败处理方式 ####

Producer的send方法本身支持内部重试，重试逻辑如下：

*  至多重试2次（同步发送为2次，异步发送为0次）。
 *  如果发送失败，则轮转到下一个Broker。这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s。
 *  如果本身向broker发送消息产生超时异常，就不会再重试。

以上策略也是在一定程度上保证了消息可以发送成功。如果业务对消息可靠性要求比较高，建议应用增加相应的重试逻辑：比如调用send同步方法发送失败时，则尝试将消息存储到db，然后由后台线程定时重试，确保消息一定到达Broker。

上述db重试方式为什么没有集成到MQ客户端内部做，而是要求应用自己去完成，主要基于以下几点考虑：首先，MQ的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是cpu、内存、网络。其次，如果MQ客户端内部集成一个KV存储模块，那么数据只有同步落盘才能较可靠，而同步落盘本身性能开销较大，所以通常会采用异步落盘，又由于应用关闭过程不受MQ运维人员控制，可能经常会发生 kill -9 这样暴力方式关闭，造成数据没有及时落盘而丢失。第三，Producer所在机器的可靠性较低，一般为虚拟机，不适合存储重要数据。综上，建议重试过程交由应用来控制。

#### 1.3选择oneway形式发送 ####

通常消息的发送是这样一个过程：

*  客户端发送请求到服务器
 *  服务器处理请求
 *  服务器向客户端返回应答

所以，一次消息发送的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用oneway形式调用，oneway形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个操作系统系统调用的开销，即将数据写入客户端的socket缓冲区，此过程耗时通常在微秒级。

### 2 消费者 ###

#### 2.1 消费过程幂等 ####

RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，例如订单Id等。在消费之前判断唯一键是否在关系数据库中存在。如果不存在则插入，并消费，否则跳过。（实际过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）

msgId一定是全局唯一标识符，但是实际使用中，可能会存在相同的消息有两个不同msgId的情况（消费者主动重发、因客户端重投机制导致的重复等），这种情况就需要使业务字段进行重复消费。

#### 2.2 消费速度慢的处理方式 ####

##### 1 提高消费并行度 #####

绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：

*  同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。
 *  提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。

##### 2 批量方式消费 #####

某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

##### 3 跳过非重要消息 #####

发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。示例代码如下：

public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) { 
            long offset = msgs.get(0).getQueueOffset();
            String maxOffset =
                    msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
            long diff = Long.parseLong(maxOffset) - offset;
            if (diff > 100000) { 
                // TODO 消息堆积情况的特殊处理
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            // TODO 正常消费过程
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

##### 4 优化每条消息消费过程 #####

举例如下，某条消息的消费过程如下：

*  根据消息从 DB 查询【数据 1】
 *  根据消息从 DB 查询【数据 2】
 *  复杂的业务计算
 *  向 DB 插入【数据 3】
 *  向 DB 插入【数据 4】

这条消息的消费过程中有4次与 DB的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把DB部署在SSD硬盘，相比于SCSI磁盘，前者的RT会小很多。

#### 2.3 消费打印日志 ####

如果消息量较少，建议在消费入口方法打印消息，消费耗时等，方便后续排查问题。

public ConsumeConcurrentlyStatus consumeMessage(
                List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) { 
            log.info("RECEIVE_MSG_BEGIN: " + msgs.toString());
            // TODO 正常消费过程
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

如果能打印每条消息消费耗时，那么在排查消费慢等线上问题时，会更方便。

#### 2.4 其他消费建议 ####

##### 1 关于消费者和订阅 #####

第一件需要注意的事情是，不同的消费者组可以独立的消费一些 topic，并且每个消费者组都有自己的消费偏移量，请确保同一组内的每个消费者订阅信息保持一致。

##### 2 关于有序消息 #####

消费者将锁定每个消息队列，以确保他们被逐个消费，虽然这将会导致性能下降，但是当你关心消息顺序的时候会很有用。我们不建议抛出异常，你可以返回 ConsumeOrderlyStatus.SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT 作为替代。

##### 3 关于并发消费 #####

顾名思义，消费者将并发消费这些消息，建议你使用它来获得良好性能，我们不建议抛出异常，你可以返回 ConsumeConcurrentlyStatus.RECONSUME\_LATER 作为替代。

##### 4 关于消费状态Consume Status #####

对于并发的消费监听器，你可以返回 RECONSUME\_LATER 来通知消费者现在不能消费这条消息，并且希望可以稍后重新消费它。然后，你可以继续消费其他消息。对于有序的消息监听器，因为你关心它的顺序，所以不能跳过消息，但是你可以返回SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT 告诉消费者等待片刻。

##### 5 关于Blocking #####

不建议阻塞监听器，因为它会阻塞线程池，并最终可能会终止消费进程

##### 6 关于线程数设置 #####

消费者使用 ThreadPoolExecutor 在内部对消息进行消费，所以你可以通过设置 setConsumeThreadMin 或 setConsumeThreadMax 来改变它。

##### 7 关于消费位点 #####

当建立一个新的消费者组时，需要决定是否需要消费已经存在于 Broker 中的历史消息CONSUME\_FROM\_LAST\_OFFSET 将会忽略历史消息，并消费之后生成的任何消息。CONSUME\_FROM\_FIRST\_OFFSET 将会消费每个存在于 Broker 中的信息。你也可以使用 CONSUME\_FROM\_TIMESTAMP 来消费在指定时间戳后产生的消息。

### 3 Broker ###

#### 3.1 Broker 角色 ####

Broker 角色分为 ASYNC\_MASTER（异步主机）、SYNC\_MASTER（同步主机）以及SLAVE（从机）。如果对消息的可靠性要求比较严格，可以采用 SYNC\_MASTER加SLAVE的部署方式。如果对消息可靠性要求不高，可以采用ASYNC\_MASTER加SLAVE的部署方式。如果只是测试方便，则可以选择仅ASYNC\_MASTER或仅SYNC\_MASTER的部署方式。

#### 3.2 FlushDiskType ####

SYNC\_FLUSH（同步刷新）相比于ASYNC\_FLUSH（异步处理）会损失很多性能，但是也更可靠，所以需要根据实际的业务场景做好权衡。

#### 3.3 Broker 配置 ####

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>listenPort</td> 
   <td>10911</td> 
   <td>接受客户端连接的监听端口</td> 
  </tr> 
  <tr> 
   <td>namesrvAddr</td> 
   <td>null</td> 
   <td>nameServer 地址</td> 
  </tr> 
  <tr> 
   <td>brokerIP1</td> 
   <td>网卡的 InetAddress</td> 
   <td>当前 broker 监听的 IP</td> 
  </tr> 
  <tr> 
   <td>brokerIP2</td> 
   <td>跟 brokerIP1 一样</td> 
   <td>存在主从 broker 时，如果在 broker 主节点上配置了 brokerIP2 属性，broker 从节点会连接主节点配置的 brokerIP2 进行同步</td> 
  </tr> 
  <tr> 
   <td>brokerName</td> 
   <td>null</td> 
   <td>broker 的名称</td> 
  </tr> 
  <tr> 
   <td>brokerClusterName</td> 
   <td>DefaultCluster</td> 
   <td>本 broker 所属的 Cluser 名称</td> 
  </tr> 
  <tr> 
   <td>brokerId</td> 
   <td>0</td> 
   <td>broker id, 0 表示 master, 其他的正整数表示 slave</td> 
  </tr> 
  <tr> 
   <td>storePathCommitLog</td> 
   <td>$HOME/store/commitlog/</td> 
   <td>存储 commit log 的路径</td> 
  </tr> 
  <tr> 
   <td>storePathConsumerQueue</td> 
   <td>$HOME/store/consumequeue/</td> 
   <td>存储 consume queue 的路径</td> 
  </tr> 
  <tr> 
   <td>mappedFileSizeCommitLog</td> 
   <td>1024 * 1024 * 1024(1G)</td> 
   <td>commit log 的映射文件大小</td> 
  </tr> 
  <tr> 
   <td>deleteWhen</td> 
   <td>04</td> 
   <td>在每天的什么时间删除已经超过文件保留时间的 commit log</td> 
  </tr> 
  <tr> 
   <td>fileReservedTime</td> 
   <td>72</td> 
   <td>以小时计算的文件保留时间</td> 
  </tr> 
  <tr> 
   <td>brokerRole</td> 
   <td>ASYNC_MASTER</td> 
   <td>SYNC_MASTER/ASYNC_MASTER/SLAVE</td> 
  </tr> 
  <tr> 
   <td>flushDiskType</td> 
   <td>ASYNC_FLUSH</td> 
   <td>SYNC_FLUSH/ASYNC_FLUSH SYNC_FLUSH 模式下的 broker 保证在收到确认生产者之前将消息刷盘。ASYNC_FLUSH 模式下的 broker 则利用刷盘一组消息的模式，可以取得更好的性能。</td> 
  </tr> 
 </tbody> 
</table>

### 4 NameServer ###

RocketMQ 中，Name Servers 被设计用来做简单的路由管理。其职责包括：

*  Brokers 定期向每个名称服务器注册路由数据。
 *  名称服务器为客户端，包括生产者，消费者和命令行客户端提供最新的路由信息。

### 5 客户端配置 ###

相对于RocketMQ的Broker集群，生产者和消费者都是客户端。本小节主要描述生产者和消费者公共的行为配置。

#### 5.1 客户端寻址方式 ####

RocketMQ可以令客户端找到Name Server, 然后通过Name Server再找到Broker。如下所示有多种配置方式，优先级由高到低，高优先级会覆盖低优先级。

*  代码中指定Name Server地址，多个namesrv地址之间用分号分割

producer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");  
    
    consumer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");

*  Java启动参数中指定Name Server地址

-Drocketmq.namesrv.addr=192.168.0.1:9876;192.168.0.2:9876

*  环境变量指定Name Server地址

export   NAMESRV_ADDR=192.168.0.1:9876;192.168.0.2:9876

*  HTTP静态服务器寻址（默认）

客户端启动后，会定时访问一个静态HTTP服务器，地址如下：[http://jmenv.tbsite.net:8080/rocketmq/nsaddr][http_jmenv.tbsite.net_8080_rocketmq_nsaddr]，这个URL的返回内容如下：

192.168.0.1:9876;192.168.0.2:9876

客户端默认每隔2分钟访问一次这个HTTP服务器，并更新本地的Name Server地址。URL已经在代码中硬编码，可通过修改/etc/hosts文件来改变要访问的服务器，例如在/etc/hosts增加如下配置：

10.232.22.67    jmenv.taobao.net

推荐使用HTTP静态服务器寻址方式，好处是客户端部署简单，且Name Server集群可以热升级。

#### 5.2 客户端配置 ####

DefaultMQProducer、TransactionMQProducer、DefaultMQPushConsumer、DefaultMQPullConsumer都继承于ClientConfig类，ClientConfig为客户端的公共配置类。客户端的配置都是get、set形式，每个参数都可以用spring来配置，也可以在代码中配置，例如namesrvAddr这个参数可以这样配置，producer.setNamesrvAddr(“192.168.0.1:9876”)，其他参数同理。

##### 1 客户端的公共配置 #####

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>namesrvAddr</td> 
   <td></td> 
   <td>Name Server地址列表，多个NameServer地址用分号隔开</td> 
  </tr> 
  <tr> 
   <td>clientIP</td> 
   <td>本机IP</td> 
   <td>客户端本机IP地址，某些机器会发生无法识别客户端IP地址情况，需要应用在代码中强制指定</td> 
  </tr> 
  <tr> 
   <td>instanceName</td> 
   <td>DEFAULT</td> 
   <td>客户端实例名称，客户端创建的多个Producer、Consumer实际是共用一个内部实例（这个实例包含网络连接、线程资源等）</td> 
  </tr> 
  <tr> 
   <td>clientCallbackExecutorThreads</td> 
   <td>4</td> 
   <td>通信层异步回调线程数</td> 
  </tr> 
  <tr> 
   <td>pollNameServerInteval</td> 
   <td>30000</td> 
   <td>轮询Name Server间隔时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>heartbeatBrokerInterval</td> 
   <td>30000</td> 
   <td>向Broker发送心跳间隔时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>persistConsumerOffsetInterval</td> 
   <td>5000</td> 
   <td>持久化Consumer消费进度间隔时间，单位毫秒</td> 
  </tr> 
 </tbody> 
</table>

##### 2 Producer配置 #####

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>producerGroup</td> 
   <td>DEFAULT_PRODUCER</td> 
   <td>Producer组名，多个Producer如果属于一个应用，发送同样的消息，则应该将它们归为同一组</td> 
  </tr> 
  <tr> 
   <td>createTopicKey</td> 
   <td>TBW102</td> 
   <td>在发送消息时，自动创建服务器不存在的topic，需要指定Key，该Key可用于配置发送消息所在topic的默认路由。</td> 
  </tr> 
  <tr> 
   <td>defaultTopicQueueNums</td> 
   <td>4</td> 
   <td>在发送消息，自动创建服务器不存在的topic时，默认创建的队列数</td> 
  </tr> 
  <tr> 
   <td>sendMsgTimeout</td> 
   <td>10000</td> 
   <td>发送消息超时时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>compressMsgBodyOverHowmuch</td> 
   <td>4096</td> 
   <td>消息Body超过多大开始压缩（Consumer收到消息会自动解压缩），单位字节</td> 
  </tr> 
  <tr> 
   <td>retryAnotherBrokerWhenNotStoreOK</td> 
   <td>FALSE</td> 
   <td>如果发送消息返回sendResult，但是sendStatus!=SEND_OK，是否重试发送</td> 
  </tr> 
  <tr> 
   <td>retryTimesWhenSendFailed</td> 
   <td>2</td> 
   <td>如果消息发送失败，最大重试次数，该参数只对同步发送模式起作用</td> 
  </tr> 
  <tr> 
   <td>maxMessageSize</td> 
   <td>4MB</td> 
   <td>客户端限制的消息大小，超过报错，同时服务端也会限制，所以需要跟服务端配合使用。</td> 
  </tr> 
  <tr> 
   <td>transactionCheckListener</td> 
   <td></td> 
   <td>事务消息回查监听器，如果发送事务消息，必须设置</td> 
  </tr> 
  <tr> 
   <td>checkThreadPoolMinSize</td> 
   <td>1</td> 
   <td>Broker回查Producer事务状态时，线程池最小线程数</td> 
  </tr> 
  <tr> 
   <td>checkThreadPoolMaxSize</td> 
   <td>1</td> 
   <td>Broker回查Producer事务状态时，线程池最大线程数</td> 
  </tr> 
  <tr> 
   <td>checkRequestHoldMax</td> 
   <td>2000</td> 
   <td>Broker回查Producer事务状态时，Producer本地缓冲请求队列大小</td> 
  </tr> 
  <tr> 
   <td>RPCHook</td> 
   <td>null</td> 
   <td>该参数是在Producer创建时传入的，包含消息发送前的预处理和消息响应后的处理两个接口，用户可以在第一个接口中做一些安全控制或者其他操作。</td> 
  </tr> 
 </tbody> 
</table>

##### 3 PushConsumer配置 #####

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>consumerGroup</td> 
   <td>DEFAULT_CONSUMER</td> 
   <td>Consumer组名，多个Consumer如果属于一个应用，订阅同样的消息，且消费逻辑一致，则应该将它们归为同一组</td> 
  </tr> 
  <tr> 
   <td>messageModel</td> 
   <td>CLUSTERING</td> 
   <td>消费模型支持集群消费和广播消费两种</td> 
  </tr> 
  <tr> 
   <td>consumeFromWhere</td> 
   <td>CONSUME_FROM_LAST_OFFSET</td> 
   <td>Consumer启动后，默认从上次消费的位置开始消费，这包含两种情况：一种是上次消费的位置未过期，则消费从上次中止的位置进行；一种是上次消费位置已经过期，则从当前队列第一条消息开始消费</td> 
  </tr> 
  <tr> 
   <td>consumeTimestamp</td> 
   <td>半个小时前</td> 
   <td>只有当consumeFromWhere值为CONSUME_FROM_TIMESTAMP时才起作用。</td> 
  </tr> 
  <tr> 
   <td>allocateMessageQueueStrategy</td> 
   <td>AllocateMessageQueueAveragely</td> 
   <td>Rebalance算法实现策略</td> 
  </tr> 
  <tr> 
   <td>subscription</td> 
   <td></td> 
   <td>订阅关系</td> 
  </tr> 
  <tr> 
   <td>messageListener</td> 
   <td></td> 
   <td>消息监听器</td> 
  </tr> 
  <tr> 
   <td>offsetStore</td> 
   <td></td> 
   <td>消费进度存储</td> 
  </tr> 
  <tr> 
   <td>consumeThreadMin</td> 
   <td>10</td> 
   <td>消费线程池最小线程数</td> 
  </tr> 
  <tr> 
   <td>consumeThreadMax</td> 
   <td>20</td> 
   <td>消费线程池最大线程数</td> 
  </tr> 
  <tr> 
   <td>consumeConcurrentlyMaxSpan</td> 
   <td>2000</td> 
   <td>单队列并行消费允许的最大跨度</td> 
  </tr> 
  <tr> 
   <td>pullThresholdForQueue</td> 
   <td>1000</td> 
   <td>拉消息本地队列缓存消息最大数</td> 
  </tr> 
  <tr> 
   <td>pullInterval</td> 
   <td>0</td> 
   <td>拉消息间隔，由于是长轮询，所以为0，但是如果应用为了流控，也可以设置大于0的值，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>consumeMessageBatchMaxSize</td> 
   <td>1</td> 
   <td>批量消费，一次消费多少条消息</td> 
  </tr> 
  <tr> 
   <td>pullBatchSize</td> 
   <td>32</td> 
   <td>批量拉消息，一次最多拉多少条</td> 
  </tr> 
 </tbody> 
</table>

##### 4 PullConsumer配置 #####

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>consumerGroup</td> 
   <td>DEFAULT_CONSUMER</td> 
   <td>Consumer组名，多个Consumer如果属于一个应用，订阅同样的消息，且消费逻辑一致，则应该将它们归为同一组</td> 
  </tr> 
  <tr> 
   <td>brokerSuspendMaxTimeMillis</td> 
   <td>20000</td> 
   <td>长轮询，Consumer拉消息请求在Broker挂起最长时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>consumerTimeoutMillisWhenSuspend</td> 
   <td>30000</td> 
   <td>长轮询，Consumer拉消息请求在Broker挂起超过指定时间，客户端认为超时，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>consumerPullTimeoutMillis</td> 
   <td>10000</td> 
   <td>非长轮询，拉消息超时时间，单位毫秒</td> 
  </tr> 
  <tr> 
   <td>messageModel</td> 
   <td>BROADCASTING</td> 
   <td>消息支持两种模式：集群消费和广播消费</td> 
  </tr> 
  <tr> 
   <td>messageQueueListener</td> 
   <td></td> 
   <td>监听队列变化</td> 
  </tr> 
  <tr> 
   <td>offsetStore</td> 
   <td></td> 
   <td>消费进度存储</td> 
  </tr> 
  <tr> 
   <td>registerTopics</td> 
   <td></td> 
   <td>注册的topic集合</td> 
  </tr> 
  <tr> 
   <td>allocateMessageQueueStrategy</td> 
   <td>AllocateMessageQueueAveragely</td> 
   <td>Rebalance算法实现策略</td> 
  </tr> 
 </tbody> 
</table>

##### 5 Message数据结构 #####

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Topic</td> 
   <td>null</td> 
   <td>必填，消息所属topic的名称</td> 
  </tr> 
  <tr> 
   <td>Body</td> 
   <td>null</td> 
   <td>必填，消息体</td> 
  </tr> 
  <tr> 
   <td>Tags</td> 
   <td>null</td> 
   <td>选填，消息标签，方便服务器过滤使用。目前只支持每个消息设置一个tag</td> 
  </tr> 
  <tr> 
   <td>Keys</td> 
   <td>null</td> 
   <td>选填，代表这条消息的业务关键词，服务器会根据keys创建哈希索引，设置后，可以在Console系统根据Topic、Keys来查询消息，由于是哈希索引，请尽可能保证key唯一，例如订单号，商品Id等。</td> 
  </tr> 
  <tr> 
   <td>Flag</td> 
   <td>0</td> 
   <td>选填，完全由应用来设置，RocketMQ不做干预</td> 
  </tr> 
  <tr> 
   <td>DelayTimeLevel</td> 
   <td>0</td> 
   <td>选填，消息延时级别，0表示不延时，大于0会延时特定的时间才会被消费</td> 
  </tr> 
  <tr> 
   <td>WaitStoreMsgOK</td> 
   <td>TRUE</td> 
   <td>选填，表示消息是否在服务器落盘后才返回应答。</td> 
  </tr> 
 </tbody> 
</table>

### 6 系统配置 ###

本小节主要介绍系统（JVM/OS）相关的配置。

#### 6.1 JVM选项 ####

推荐使用最新发布的JDK 1.8版本。通过设置相同的Xms和Xmx值来防止JVM调整堆大小以获得更好的性能。简单的JVM配置如下所示：  
  
` -server -Xms8g -Xmx8g -Xmn4g `  
  
  
如果您不关心RocketMQ Broker的启动时间，还有一种更好的选择，就是通过“预触摸”Java堆以确保在JVM初始化期间每个页面都将被分配。那些不关心启动时间的人可以启用它：  
 -XX:+AlwaysPreTouch  
禁用偏置锁定可能会减少JVM暂停，  
 -XX:-UseBiasedLocking  
至于垃圾回收，建议使用带JDK 1.8的G1收集器。

-XX:+UseG1GC -XX:G1HeapRegionSize=16m   
    -XX:G1ReservePercent=25 
    -XX:InitiatingHeapOccupancyPercent=30

这些GC选项看起来有点激进，但事实证明它在我们的生产环境中具有良好的性能。另外不要把-XX:MaxGCPauseMillis的值设置太小，否则JVM将使用一个小的年轻代来实现这个目标，这将导致非常频繁的minor GC，所以建议使用rolling GC日志文件：

-XX:+UseGCLogFileRotation   
    -XX:NumberOfGCLogFiles=5 
    -XX:GCLogFileSize=30m

如果写入GC文件会增加代理的延迟，可以考虑将GC日志文件重定向到内存文件系统：

-Xloggc:/dev/shm/mq_gc_%p.log123

#### 6.2 Linux内核参数 ####

os.sh脚本在bin文件夹中列出了许多内核参数，可以进行微小的更改然后用于生产用途。下面的参数需要注意，更多细节请参考/proc/sys/vm/\*的[文档][Link 1]

*  **vm.extra\_free\_kbytes**，告诉VM在后台回收（kswapd）启动的阈值与直接回收（通过分配进程）的阈值之间保留额外的可用内存。RocketMQ使用此参数来避免内存分配中的长延迟。（与具体内核版本相关）
 *  **vm.min\_free\_kbytes**，如果将其设置为低于1024KB，将会巧妙的将系统破坏，并且系统在高负载下容易出现死锁。
 *  **vm.max\_map\_count**，限制一个进程可能具有的最大内存映射区域数。RocketMQ将使用mmap加载CommitLog和ConsumeQueue，因此建议将为此参数设置较大的值。（agressiveness --> aggressiveness）
 *  **vm.swappiness**，定义内核交换内存页面的积极程度。较高的值会增加攻击性，较低的值会减少交换量。建议将值设置为10来避免交换延迟。
 *  **File descriptor limits**，RocketMQ需要为文件（CommitLog和ConsumeQueue）和网络连接打开文件描述符。我们建议设置文件描述符的值为655350。
 *  [Disk scheduler][]，RocketMQ建议使用I/O截止时间调度器，它试图为请求提供有保证的延迟。

[architecture]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzA2MDE2NS5wbmc?x-oss-process=image/format,png
[architecture 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzI0MTIwNC5wbmc?x-oss-process=image/format,png
[architecture 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzI4Mzk1Ny5wbmc?x-oss-process=image/format,png
[design]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzMyNTIwNC5wbmc?x-oss-process=image/format,png
[design 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzM1NzQ5Ni5wbmc?x-oss-process=image/format,png
[design 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzM4NzA3Ny5wbmc?x-oss-process=image/format,png
[design 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzQxMDM2NC5wbmc?x-oss-process=image/format,png
[design 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzQ0NTA2My5wbmc?x-oss-process=image/format,png
[design 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzQ3MDIyNC5wbmc?x-oss-process=image/format,png
[design 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzU0MzIxNy5wbmc?x-oss-process=image/format,png
[design 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MjkxNjc3MS5wbmc?x-oss-process=image/format,png
[design 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzU5NDQ3MC5wbmc?x-oss-process=image/format,png
[design 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzYzMTA0Ni5wbmc?x-oss-process=image/format,png
[design 10]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzY1MDkxNy5wbmc?x-oss-process=image/format,png
[design 11]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzY3Nzk3NC5wbmc?x-oss-process=image/format,png
[design 12]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3l6LWltYWdlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vVVRPT0xTMTU3ODg3MzcwNDMzNC5wbmc?x-oss-process=image/format,png
[OpenMessaging]: https://www.google.com/url?q=http://openmessaging.cloud/&sa=D&ust=1546524111089000
[http_jmenv.tbsite.net_8080_rocketmq_nsaddr]: http://jmenv.tbsite.net:8080/rocketmq/nsaddr
[Link 1]: https://www.kernel.org/doc/Documentation/sysctl/vm.txt
[Disk scheduler]: https://access.redhat.com/documentation/en-US/Red_Hat_Enterprise_Linux/6/html/Performance_Tuning_Guide/ch06s04s02.html