# RocketMQ集群与运维 ## 1. 集群搭建方式 集群模式: - 单Master - 多Master **优点**：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高； **缺点**：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。 - 多Master多Slave模式(异步) 每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级） **优点**：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样； **缺点**：Master宕机，磁盘损坏情况下会丢失少量消息。 - 多Master 多Slave模式(同步) 每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功,才向应用返回成功 **优点**：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高； **缺点**：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。 ## 2. 集群的搭建 ### 2.1 前置配置 参考第1节安装教程 ### 2.2 单Master模式 参考第1节 ### 2.3 多Master模式 **rocketmq /conf目录下提供了以下几种类型的配置** - 启动name server ```sh ### 首先启动NameServer $ nohup sh mqnamesrv & ### 检查NameServer是否启动成功 $ tail -f ~/logs/rocketmqlogs/namesrv.log The Name Server boot success... ``` - 启动Broker集群 ```sh #注意配置文件根据自己机器路径取就行 ### 启动第一个broker，假定namesrv在192.168.1.1上。注意这里的配置文件的位置 $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties & ### 在第二台服务器上启动另一个broker。 $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties & ``` 上面的NameServer是一台，IP地址直接写，如果是多台NameServer，则需要在 -n 后接多个NameServer的地址，使用分号分开。由于shell对分号敏感，可以使用单引号引起来多个NameServer的地址，禁止shell对分号的解释 ### 2.4 多Master和Slave模式-异步复制 - 启动name server ```sh ### 首先启动NameServer $ nohup sh mqnamesrv & ``` - 启动broker集群 ```sh ### 启动Master的broker：broker-a $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties & ### 在另外一台服务器上启动另一个Master的broker：broker-b $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties & ### 在另一台服务器上启动broker-a的Slave $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s- async/broker-a-s.properties & ### 在另一台服务器上启动broker-b的Slave $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties & ``` ### 2.5多Master和多Slave模式-同步双写 - 启动name server ```sh ### 首先启动NameServer $ nohup sh mqnamesrv & ``` - 启动broker集群 ```sh ### 启动Master的broker：broker-a $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties & ### 在另外一台服务器上启动另一个Master的broker：broker-b $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties & ### 在另一台服务器上启动broker-a的同步Slave $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s- sync/broker-a-s.properties & ### 在另一台服务器上启动broker-b的同步Slave $ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties & ``` ## 3. mqadmin管理工具 ### 3.1 topic相关 - updateTopic: 创建更新topic配置 - -b broker地址,只支持单台 - -c cluster名称,表示topic所在集群 - -h 打印帮助 - -n name server地址 - -p 指定新topic的读写权限 - -r 可读队列数, 默认8 - -w 可写队列数,默认8 - -t topic名称 - deleteTopic 删除topic - -c cluster名称 - -n nameserver 地址 - -t top名称 - topicList 查看topic列表信息 - topicRoute 查看 Topic 路由信息 - topicStatus 查看 Topic 消息队列offset - topicClusterList 查看 Topic 所在集群列表 - updateTopicPerm 更新 Topic 读写权限 - updateOrderConf 从NameServer上创建、删除、获取特定命名空间的kv配置，目前还未启用 - allocateMQ 以平均负载算法计算消费者列表负载消息队列的负载结果 - statsAll 打印Topic订阅关系 TPS、积累量、24h读写总量等信息 ```sh #例 # 查看指定NameServer下的主题 [root@node4 ~]# mqadmin topicList -n node1:9876 # 查看指定NameServer，指定集群名称下的主题 [root@node4 ~]# mqadmin topicList -n node1:9876 -c DefaultCluster # 创建主题，指定NameServer，指定Broker 指定主题名称，指定主题的写队列个数， 指定读主题队 列个数 [root@node4 ~]# mqadmin updateTopic -b node1:10911 -r 3 -w 3 -t tp_admin_01 # 描述主题，指定NameServer，指定主题名称 [root@node4 ~]# mqadmin topicStatus -t tp_admin_01 -n node1:9876 # 创建主题，指定NameServer，指定集群名称，指定主题名称，指定读主题队列个数， 指定写主题队 列个数 [root@node4 ~]# mqadmin updateTopic -c DefaultCluster -n node1:9876 -r 3 -w 3 -t tp_admin_02 # 查看指定主题的状态，指定NameServer地址，指定主题名称 [root@node4 ~]# mqadmin topicStatus -t tp_admin_02 -n node1:9876 #删除主题，指定NameServer地址，指定集群名称，指定主题名称 [root@node3 ~]# mqadmin deleteTopic -n node1:9876 -c DefaultCluster -t tp_admin_03 # 查看主题所在的集群，指定NameServer地址，指定主题名称。因为不同集群可以拥有同名的主题， 并且不同集群可以注册到同一个NameServer [root@node3 ~]# mqadmin topicClusterList -n node1:9876 -t tp_admin_02 # 计算消费的负载均衡，不同的-i列表，计算不同的消费平衡负载结果 [root@node3 ~]# mqadmin allocateMQ -n node1:9876 -t tp_admin_02 -i node1,node3 [root@node3 ~]# mqadmin allocateMQ -n node1:9876 -t tp_admin_02 -i node1,node2,node3,node4 # 打印Topic订阅关系、TPS、积累量、24h读写总量等信息 [root@node3 ~]# mqadmin statsAll -n node1:9876 ``` ### 3.2 集群相关 - clusterList 查看集群信息 - clusterRT 发送消息检测集群各BrokerRT。消息发往${BrokerName} Topic。 ### 3.3 Broker相关 - updateBrokerConfig 更新 Broker 配置文件，会修改Broker.conf - brokerStatus 查看 Broker 统计信息、运行状态 - brokerConsumeStats Broker中各个消费者的消费情况，按Message Queue维度返回Consume Offset，Broker Offset，Diff，TImestamp等信息 - getBrokerConfig 获取Broker配置 - wipeWritePerm 从NameServer上清除 Broker写权限 - cleanExpiredCQ 清理Broker上过期的Consume Queue，如果手动减少对列数可能产生过期队列 - cleanUnusedTopic 清理Broker上不使用的Topic - sendMsgStatus 向Broker发消息，返回发送状态和RT ### 3.4 消息相关 - queryMsgById 根据offsetMsgId查询msg - queryMsgByKey - queryMsgByOffset - queryMsgByUniqueKey 根据msgId查询 - checkMsgSendRT 检测向topic发消息的RT - sendMessage - consumeMessage - printMsg - printMsgByQueue 类似printMsg，但指定Message Queue - resetOffsetByTime 按时间戳重置offset，Broker和consumer ### 3.5 消费者/消费组相关 - consumerProgress 查看订阅组消费状态 - consumerStatus - updateSubGroup 更新或创建订阅关系 - deleteSubGroup 从Broker删除订阅关系 - cloneGroupOffset 在目标群组中使用源群组的offset ### 3.6 连接相关 - consumerConnection 查询 Consumer 的网络连接 - producerConnection 查询 Producer 的网络连接 ### 3.7 NameServer相关 - updateKvConfig 更新NameServer的kv配置 - deleteKvConfig 删除NameServer的kv配置 - getNamesrvConfig 获取NameServer配置 - updateNamesrvConfig 修改NameServer配置 ### 3.8 其他 - startMonitoring 开启监控进程，监控消息误删、重试队列消息数等 - ...... ## 4. 运维常见问题 ### 4.1 mqadmin 命令报错问题 ``` org.apache.rocketmq.remoting.exception.RemotingConnectException:connectto failed ``` 解决方法：可以在部署RocketMQ集群的虚拟机上执行export NAMESRV_ADDR=ip:9876 ### 4.2 生产端和消费端版本不一致导致不能正常消费的问题 解决方案：RocketMQ 的jar包：rocketmq-client等包应该保持生产端，消费端使用相同的version ### 4.3 新增一个topic消费组时, 无法消费历史消息的问题 解决方案：rocketmq默认策略是从消息队列尾部，即跳过历史消息。如果想消费历史消息，则需要设置 ```properties .apache.rocketmq.client.consumer.DefaultMQPushConsumer#setConsumeFromWhere # 1.默认,一个新的订阅组第一次启动从队列的最后位置开始消费，后续再启动接着上次消费的进度开始消费,即跳过历史消息； # 2.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);一个新的订阅组第一次启动从队列的最前位置开始消费，后续再启动接着上次消费的进度开始消费,即消费Broker未过期的历史消息； # 3.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);一个新的订阅组第一次启动从指定时间点开始消费，后续再启动接着上次消费的进度开始消费，和consumer.setConsumeTimestamp()配合使用，默认是半个小时以前； # 4.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); ``` ### 4.4 如何开启从slave读数据功能 在某些情况下，Consumer需要将消费位点重置到1-2天前，这时在内存有限的Master Broker上，CommitLog会承载比较重的IO压力，影响到该Broker的其它消息的读与写。可以开启**slaveReadEnable=true** ，当Master Broker发现Consumer的消费位点与CommitLog的最新值的差值的容量超过该机器内存的百分比（ accessMessageInMemoryMaxRatio=40% ），会推荐Consumer从Slave Broker中去读取数据，降低Master Broker的IO。 ### 4.5 性能调优问题 异步刷盘建议使用**自旋锁**，同步刷盘建议使用**重入锁**，调整Broker配置项useReentrantLockWhenPutMessage ，默认为false； 异步刷盘建议开启TransientStorePoolEnable ；建议关闭transferMsgByHeap，提高拉消息效率； 同步刷盘建议适当增大 sendMessageThreadPoolNums ，具体配置需要经过压测 ### 4.6 msgId和offsetMsgid含义与区别 - msgId，对于客户端来说msgId是由客户端producer实例端生成的，具体来说，调用以下方法生成唯一的Id； ``` MessageClientIDSetter.createUniqIDBuffer() ``` - offsetMsgId，offsetMsgId是由Broker服务端在写入消息时生成的（采用”IP地址+Port端口” 与“CommitLog的物理偏移量地址”做了一个字符串拼接），其中offsetMsgId就是在RocketMQ控制台直接输入查询的那个messageId。