RocketMQ最佳实践

蔚落 2022-11-18 01:51 228阅读 0赞

技术活，该赏
    关注+一键三连（点赞，评论，收藏）再看，养成好习惯

[RocketMQ使用教程相关系列 目录][RocketMQ_]

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

# 最佳实践 #

介绍RocketMQ的最佳实践，包括生产者、消费者、Broker以及NameServer的最佳实践，客户端的配置方式以及JVM和linux的最佳参数配置。

## 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>mapedFileSizeCommitLog</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>fileReserverdTime</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></td> 
   <td></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截止时间调度器，它试图为请求提供有保证的延迟。  
      
    参考：https://github.com/apache/rocketmq/tree/master/docs/cn

[RocketMQ_]: https://blog.csdn.net/shi_hong_fei_hei/article/details/115470425
[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