一、RocketMQ简介:
RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点：
1、支持严格的消息顺序；
2、支持Topic与Queue两种模式；
3、亿级消息堆积能力；
4、比较友好的分布式特性；
5、同时支持Push与Pull方式消费消息；
官网链接:
rocketmq下载地址： https://github.com/alibaba/RocketMQ/releases
rocketmq github: https://github.com/alibaba/RocketMQ
RocketMQ集群安装：
RocketMQ有多种集群的方式，这里是双master的集群.
优点：优点：配置简单，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢）性能最高
一、前期准备
主机（两台）为centOs6.5系统,ip对应主机名如下：
10.10.10.23 rocketmq-master1 #ip和主机名 10.10.10.24 rocketmq-master2
所需软件：
jdk1..0_45.tar.gz #jdk软件 alibaba-rocketmq-.tar.gz #mq软件包
hosts信息添加：(两台主机均的添加hosts)
# vim /etc/hosts 10.10.10.23 mqnameserver1 #ip对应brokerName 10.10.10.24 mqnameserver2 10.10.10.23 rocketmq-master1 #ip对应主机名 10.10.10.24 rocketmq-master2 #ip对应主机名
二、安装：
1.>安装JDK
jdk1.6.0_45.tar.gz #软件包 #tar -zxvf jdk1.6.0_45.tar.gz -C /apps/product/ #ln -s /apps/product/jdk1.6.0_45 java #环境变量配置 #vim /etc/profile JAVA_HOME=/apps/product/java CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar PATH=$JAVA_HOME/bin:$PATH export JAVA_HOME CLASSPATH PATH source /etc/profile 查看jdk版本：java -version
2.>RocketMQ安装: (mq版本：3.2.6)
>>tar -zxvf alibaba-rocketmq-3.2.6.tar.gz -C /apps/product/ >>cd /apps/product/ >>mv alibaba-rocketmq-3.2.6 rocketmq 环境变量设置: >>vim /etc/profile JAVA_HOME=/apps/product/java CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar PATH=$JAVA_HOME/bin:$PATH ROCKETMQ_HOME=/apps/product/rocketmq export JAVA_HOME CLASSPATH PATH ROCKETMQ_HOME >>source /etc/profile >>useradd admin >>cd /apps/product/rocketmq/bin && sh os.sh
3.RocketMQ配置:
部署Broker：消息中转角色，负责存储消息，转发消息
Broker配置参数:
1.>获取Broker的默认配置
sh mqbroker -m
2.>Broker启动时，如何加载配置
生成Broker默认配置模版
sh mqbroker -m >broker.p
3.>修改配置文件.
broker.p
4.>加载修改过的配置文件
nohup sh mqbroker -c broker.p #将broker.p文件复制到broker-a.properties文件，并可以添加其他配置.
(1)、Master1服务器(10.10.10.23)
vim /apps/product/rocketmq/conf/2m-noslave/broker-a.properties rocketmqHome=/apps/product/rocketmq namesrvAddr=10.10.10.23:9876;10.10.10.24:9876 brokerIP1=10.10.10.23 brokerName=rocketmq-master1 brokerClusterName=DefaultCluster brokerId=0 autoCreateTopicEnable=false autoCreateSubscriptionGroup=true rejectTransactionMessage=false fetchNamesrvAddrByAddressServer=false storePathRootDir=/root/store storePathCommitLog=/root/store/commitlog flushIntervalCommitLog=1000 flushCommitLogTimed=false deleteWhen=04 fileReservedTime=72 diskMaxUsedSpaceRatio=88 maxTransferBytesOnMessageInMemory=262144 maxTransferCountOnMessageInMemory=32 maxTransferBytesOnMessageInDisk=65536 maxTransferCountOnMessageInDisk=8 accessMessageInMemoryMaxRatio=40 messageIndexEnable=true messageIndexSafe=false haMasterAddress= brokerRole=ASYNC_MASTER flushDiskType=ASYNC_FLUSH cleanFileForciblyEnable=true sendMessageThreadPoolNums=128 pullMessageThreadPoolNums=128 brokerPermission=6 defaultTopicQueueNums=8 clusterTopicEnable=true brokerTopicEnable=true adminBrokerThreadPoolNums=16 clientManageThreadPoolNums=16 flushConsumerOffsetInterval=5000 flushConsumerOffsetHistoryInterval=60000 sendThreadPoolQueueCapacity=100000 pullThreadPoolQueueCapacity=100000 filterServerNums=0 longPollingEnable=true shortPollingTimeMills=1000 notifyConsumerIdsChangedEnable=true offsetCheckInSlave=false listenPort=10911 serverWorkerThreads=8 serverCallbackExecutorThreads=0 serverSelectorThreads=3 serverOnewaySemaphoreValue=256 serverAsyncSemaphoreValue=64 serverChannelMaxIdleTimeSeconds=120 serverSocketSndBufSize=131072 serverSocketRcvBufSize=131072 serverPooledByteBufAllocatorEnable=false clientWorkerThreads=4 clientCallbackExecutorThreads=2 clientOnewaySemaphoreValue=2048 clientAsyncSemaphoreValue=2048 connectTimeoutMillis=3000 channelNotActiveInterval=60000 clientChannelMaxIdleTimeSeconds=120 clientSocketSndBufSize=131072 clientSocketRcvBufSize=131072 clientPooledByteBufAllocatorEnable=false mapedFileSizeCommitLog=1073741824 mapedFileSizeConsumeQueue=6000000 flushIntervalConsumeQueue=1000 cleanResourceInterval=10000 deleteCommitLogFilesInterval=100 deleteConsumeQueueFilesInterval=100 destroyMapedFileIntervalForcibly=120000 redeleteHangedFileInterval=120000 diskMaxUsedSpaceRatio=75 putMsgIndexHightWater=600000 maxMessageSize=524288 checkCRCOnRecover=true flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueThoroughInterval=60000 maxHashSlotNum=5000000 maxIndexNum=20000000 maxMsgsNumBatch=64 haListenPort=10912 haSendHeartbeatInterval=5000 haHousekeepingInterval=20000 haTransferBatchSize=32768 haSlaveFallbehindMax=268435456 syncFlushTimeout=5000 messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h flushDelayOffsetInterval=10000
broker-a.properties
(2)、Master2服务器(10.10.10.24)
vim /apps/product/rocketmq/conf/2m-noslave/broker-b.properties rocketmqHome=/apps/product/rocketmq namesrvAddr=10.10.10.24:9876;10.10.10.23:9876 brokerIP1=10.10.10.24 brokerName=rocketmq-master2 brokerClusterName=DefaultCluster brokerId=0 autoCreateTopicEnable=false autoCreateSubscriptionGroup=true rejectTransactionMessage=false fetchNamesrvAddrByAddressServer=false storePathRootDir=/root/store storePathCommitLog=/root/store/commitlog flushIntervalCommitLog=1000 flushCommitLogTimed=false deleteWhen=04 fileReservedTime=72 diskMaxUsedSpaceRatio=88 maxTransferBytesOnMessageInMemory=262144 maxTransferCountOnMessageInMemory=32 maxTransferBytesOnMessageInDisk=65536 maxTransferCountOnMessageInDisk=8 accessMessageInMemoryMaxRatio=40 messageIndexEnable=true messageIndexSafe=false haMasterAddress= brokerRole=ASYNC_MASTER flushDiskType=ASYNC_FLUSH cleanFileForciblyEnable=true sendMessageThreadPoolNums=128 pullMessageThreadPoolNums=128 brokerPermission=6 defaultTopicQueueNums=8 clusterTopicEnable=true brokerTopicEnable=true adminBrokerThreadPoolNums=16 clientManageThreadPoolNums=16 flushConsumerOffsetInterval=5000 flushConsumerOffsetHistoryInterval=60000 sendThreadPoolQueueCapacity=100000 pullThreadPoolQueueCapacity=100000 filterServerNums=0 longPollingEnable=true shortPollingTimeMills=1000 notifyConsumerIdsChangedEnable=true offsetCheckInSlave=false listenPort=10911 serverWorkerThreads=8 serverCallbackExecutorThreads=0 serverSelectorThreads=3 serverOnewaySemaphoreValue=256 serverAsyncSemaphoreValue=64 serverChannelMaxIdleTimeSeconds=120 serverSocketSndBufSize=131072 serverSocketRcvBufSize=131072 serverPooledByteBufAllocatorEnable=false clientWorkerThreads=4 clientCallbackExecutorThreads=2 clientOnewaySemaphoreValue=2048 clientAsyncSemaphoreValue=2048 connectTimeoutMillis=3000 channelNotActiveInterval=60000 clientChannelMaxIdleTimeSeconds=120 clientSocketSndBufSize=131072 clientSocketRcvBufSize=131072 clientPooledByteBufAllocatorEnable=false mapedFileSizeCommitLog=1073741824 mapedFileSizeConsumeQueue=6000000 flushIntervalConsumeQueue=1000 cleanResourceInterval=10000 deleteCommitLogFilesInterval=100 deleteConsumeQueueFilesInterval=100 destroyMapedFileIntervalForcibly=120000 redeleteHangedFileInterval=120000 diskMaxUsedSpaceRatio=75 putMsgIndexHightWater=600000 maxMessageSize=524288 checkCRCOnRecover=true flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueThoroughInterval=60000 maxHashSlotNum=5000000 maxIndexNum=20000000 maxMsgsNumBatch=64 haListenPort=10912 haSendHeartbeatInterval=5000 haHousekeepingInterval=20000 haTransferBatchSize=32768 haSlaveFallbehindMax=268435456 syncFlushTimeout=5000 messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h flushDelayOffsetInterval=10000
broker-b.properties
配置解析注释文件：
## Broker对外服务的监听端口 listenPort = 10911 ## Name Server地址 namesrvAddr=192.168.3.229:9876 ##本机IP地址，默认系统自动识别，但是某些多网卡机器会存在识别错误的情况，这种情况下可以人工配置 brokerIP1=192.168.3.229 ##本机主机名 brokerName=dev10.com ##Broker所属哪个集群,其中缺省值为DefaultCluster brokerClusterName=DefaultCluster ## BrokerId，必须是大等于0的整数，0表示Master，>0表示Slave，一个Master可以挂多个Slave，Master与Slave通过BrokerName来配对 brokerId=0 ##是否允许Broker自动创建Topic，建议线下开启，线上关闭 autoCreateTopicEnable=true //线上改为false ##是否允许Broker自动创建订阅组，建议线下开启，线上关闭 autoCreateSubscriptionGroup=true //线上改为false ##是否拒绝事务消息接入 rejectTransactionMessage=false ##是否从web服务器获取Name Server地址，针对大规模的Broker集群建议使用这种方式 fetchNamesrvAddrByAddressServer=false ## commitLog存储路径 $HOME/store/commitlog storePathCommitLog=/root/store/commitlog ##消费队列存储路径 $HOME/store/consumequeue storePathConsumeQueue=/root/store/consumequeue ##消息索引存储路径 $HOME/store/index storePathIndex=/root/store/index ## checkpoint文件存储路径 $HOME/store/checkpoint storeCheckpoint=/root/store/checkpoint ##异常退出产生的文件-//abort文件存储路径 $HOME/store/abort abortFile=/root/store/abort ##删除文件时间点，默认凌晨4点 deleteWhen=04 ##文件保留时间，默认48小时 fileReservedTime=72 ##单次Pull消息（内存）传输的最大字节数 maxTransferBytesOnMessageInMemory=262144 ##单次Pull消息（内存）传输的最大条数 maxTransferCountOnMessageInMemory=32 ##单次Pull消息（磁盘）传输的最大字节数 maxTransferBytesOnMessageInDisk=65536 ##单次Pull消息（磁盘）传输的最大条数 maxTransferCountOnMessageInDisk=8 ##命中消息在内存的最大比例 accessMessageInMemoryMaxRatio=40 ##是否开启消息索引功能 messageIndexEnable=true ##是否提供安全的消息索引机制，索引保证不丢 messageIndexSafe=false ⇒ 线上改为true ##在Slave上直接设置Master地址，默认从Name Server上自动获取，也可以手工强制配置 haMasterAddress= ## Broker的角色 - ASYNC_MASTER异步复制Master-SYNC_MASTER同步双写Master-SLAVE brokerRole=ASYNC_MASTER ##刷盘方式 - ASYNC_FLUSH 异步刷盘 - SYNC_FLUSH 同步刷盘 flushDiskType=ASYNC_FLUSH ##磁盘满、且无过期文件情况下 TRUE 表示强制删除文件，优先保证服务可用 FALSE 标记服务不可用，文件不删除 cleanFileForciblyEnable=true ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~非默认配置~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ##服务器服务权限 默认值6 6 读写 4 只读 2只写 brokerPermission=6 ##默认8 ,topic的queue数 defaultTopicQueueNums=4 ##默认 true 自动创建以集群名字命名的Topic功能 clusterTopicEnable=true ##发送消息线程数 32+cpu*4 sendMessageThreadPoolNums=16 ##拉消息处理线程数 32+cpu*4 pullMessageThreadPoolNums=16 ##管控命令处理线程数 8 adminBrokerThreadPoolNums=8 ##Topic持久化路径 $HOME/store/config/topics.json topicConfigPath=/root/store/config/topics.json ##ConsumerOffset $HOME /store/config/topics.json consumerOffsetPath=/root/store/config/consumerOffset.json ##Broker 配置文件路径 $HOME /store/config/broker.properties #brokerConfigPath=/root/store/config /broker.properties ##subscriptionGroup 持久化路径 $HOME /store/config/topics.json subscriptionGroupPath=/root/store/config /subscriptionGroup.json ##刷消费进度 时间 默认5秒 flushConsumerOffsetInterval=5000 ##消费进度历史统计 60秒 flushConsumerOffsetHistoryInterval=60000 ##查询消息最大时间跨度，单位小时 默认3小时 queryMessageMaxTimeSpan=3 =================== NettyServerConfig=================== ##netty worker线程数默认 32 serverWorkerThreads=32 ##CallbackExecutor 默认是0 ,0表示系统值 serverCallbackExecutorThreads=0 ##netty selector 线程 默认 8 serverSelectorThreads=8 ##Oneway方式处理线程 serverOnewaySemaphoreValue=32 ##Async方式处理线程 serverAsyncSemaphoreValue=64 ##通信层最大空闲时间 默认120秒 serverChannelMaxIdleTimeSeconds=120 ==================== NettyClientConfig===================== ##clientWorker 线程 4 clientWorkerThreads=4 ##CallbackExecutor clientCallbackExecutorThreads=4 ###clientSelector 1 clientSelectorThreads=1 ##clientOneway 默认 256 事务会查时可以多些 clientOnewaySemaphoreValue=56 ##clientAsync 默认 128 暂时无太多用途 clientAsyncSemaphoreValue=28 ##连接超时时间 3秒 connectTimeoutMillis=3000 ##channel 不活动时间 无心跳时间 6秒 channelNotActiveInterval=60000 ##channel 空闲时间 120秒 clientChannelMaxIdleTimeSeconds=120 =============== MessageStoreConfig=========================== ##CommitLog每个文件大小 1G mapedFileSizeCommitLog=1073741824 ##ConsumeQueue每个文件大小 默认存储50W条消息*8 mapedFileSizeConsumeQueue=10000000 ##CommitLog刷盘间隔时间（单位毫秒） flushIntervalCommitLog=1000 ##ConsumeQueue刷盘间隔时间（单位毫秒） flushIntervalConsumeQueue=1000 ##清理资源间隔时间（单位毫秒）10秒 cleanResourceInterval=10000 ##删除多个CommitLog文件的间隔时间（单位毫秒） deleteCommitLogFilesInterval=100 ##删除多个ConsumeQueue文件的间隔时间（单位毫秒） deleteConsumeQueueFilesInterval=100 ##强制删除文件间隔时间（单位毫秒）120秒 destroyMapedFileIntervalForcibly=120000 ##定期检查Hanged文件间隔时间（单位毫秒）120秒 redeleteHangedFileInterval=120000 ##磁盘空间最大使用率75% diskMaxUsedSpaceRatio=75 ##写消息索引到ConsumeQueue，缓冲区高水位，超过则开始流控 600000 putMsgIndexHightWater=600000 ##最大消息大小，默认512K 1M maxMessageSize=1050000 ##重启时，是否校验CRC 默认是true checkCRCOnRecover=true ##刷CommitLog，至少刷几个PAGE flushCommitLogLeastPages=4 ##刷ConsumeQueue，至少刷几个PAGE flushConsumeQueueLeastPages=2 ##刷CommitLog，彻底刷盘间隔时间 10秒 flushCommitLogThoroughInterval=10000 ##刷ConsumeQueue，彻底刷盘间隔时间 60秒 flushConsumeQueueThoroughInterval=60000 ##最大被拉取的消息个数，消息在内存 ？ maxTransferCountOnMessageInMemory=32 maxHashSlotNum=5000000 maxIndexNum=20000000 最大key查找拉取条数 maxMsgsNumBatch=32 ========================集群配置部分，非集群不添加==================== ###HA功能 端口号----非集群不加该配置 haListenPort=10912 ##HA心跳间隔 5秒 haSendHeartbeatInterval=5000 ###HA保持时间20秒 haHousekeepingInterval=20000 ###HA最大一次传输 32k haTransferBatchSize=32768 ###如果不设置，则从NameServer获取Master HA服务地址 #haMasterAddress= ###Slave落后Master超过此值，则认为存在异常 1024 * 1024 * 256 256M #haSlaveFallbehindMax=268435456
Broker参数解析
(3)、runbroker.sh参数调整
vim /apps/product/rocketmq/bin/runbroker.sh runbroker.sh需要根据内存大小进行适当地调整 JAVA_OPT_1="-server-Xms8g -Xmx8g -Xmn2g -XX:PermSize=1g -XX:MaxPermSize=1g"
服务启动:
mkdir -p /data/rocketmq/store/commitlog /data/logs #创建日志存放目录
>>cd /apps/product/rocketmq/conf && sed -i 's#${user.home}#/data#g' *.xml #更改为data目录
(1)、启动NameServer【两台启动方式相同】
> cd /apps/product/rocketmq/bin > nohup sh mqnamesrv & 或者 nohup sh mqnamesrv > nohup.out 2>&1 &
(2)、启动BrokerServer A【10.10.10.23主机】
cd /apps/product/rocketmq/bin nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties >/dev/null 2>&1 & 或者 #nohup sh mqbroker -n 10.10.10.23:9876 -c ../conf/2m-noslave/broker-a.properties >/dev/null 2>&1 &
(3)、启动BrokerServer B【10.10.10.24主机】
> cd /apps/product/rocketmq/bin nohup sh mqbroker -c ../conf/2m-noslave/broker-b.properties >/dev/null 2>&1 & # nohup sh mqbroker -n 10.10.10.24:9876 -c ../conf/2m-noslave/broker-b.properties >/dev/null 2>&1 &
(4)、测试：
两台机器分别查看日志信息： tail -100f /data/logs/rocketmqlogs/namesrv.log
tail -100f /data/logs/rocketmqlogs/broker.log
#netstat -ntlp #查看端口号
# jps #查看服务
后台启动并生产日志： nohup sh mqnamesrv >& /var/log/ns.log