RocketMq消息队列实施专项方案.docx

消息队列实施方案
1、背景
异步解耦合、给前端系统提供最高效反应
2、常见消息队列对比
2、1 ActiveMq
ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范 JMS Provider实现
优点：
Java语言
支持集群模式
缺点：
性能在消息中间件中处于下游

2、2 Rabbitmq
Rabbitmq是基于AMQP使用erlang语言实现消息队列系统

优点：
1、完整消息队列系统，支持多个消息队列模式，包含竞争消费；
2、支持集群模式，扩展集群容量和性能比较方便，集成了集群监控和管理；
3、支持消息持久化；

缺点：
1、需要学习比较复杂接口和协议，比较花费时间；
2、性能不是尤其理想大约在1wqps左右；
3、使用Erlang语言，语言基础；

2、3 Kafka
Kafka 是LinkedIn 开发一个高性能、分布式消息公布订阅系统。

优点：
1、分布式集群能够透明扩展，增加新服务器进集群。
2、高性能。单机写入TPS约在百万条/秒
3、容错。数据全部会复制到几台服务器上。
缺点：
1、复杂性。Kafka需要zookeeper 集群支持，Topic通常需要人工来创建，布署和维护较通常消息队列成本更高
定在日志传输、存在消息丢失肯能、消息乱序
3、消息发送错误无重试
2、4 RocketMQ
RockerMq 是阿里企业中间件团体参考Kafka思想，用Java语言实现消息传输系统
优点：
1、较高性能，单机写入TPS单实例约7万条/秒
2、容错，多个集群模式、能够处理容错问题
3、消息重试发送
4、次序消息能够严格实施
缺点：
1、消息反复、消费端需要做去重操作
2、5 选择结论
从项目业务和团体技术偏向考虑，我们应该需要一个数据安全性比较高，确保每个消息全部会被实施；有容错机制、支持集群模式高可用；性能不错，能够在毫秒级处理消息；支持次序消息消息中间件， RockerMq 能够满足这些要求。

3、RockerMq介绍
3、1 RockerMq 产品介绍
参考阿里企业提供《RocketMQ 开发指南》，最新版针对 v3.2.4

3、2 RockerMq集群
3、2、1 布署方法
Rockermq共有四种布署方法，分别是：
1、单个 Master
一旦 Broker 重启或宕机时，会造成整个服务不可用
2、多 Master 模式
一个集群无 Slave，全是 Master，比如 2 个 Master 戒者 3 个 Master
优点：
1、配置简单，
2、容错，单个 Master 宕机或重启维护对应用无影响，在磁盘配置为 RAID10 时，即使机器宕机不可恢复情况下，因为 RAID10 磁盘很可靠，在同时刷盘时消息不会丢，异步刷盘丢失少许消息，
3、性能最高。
3、多 Master 多 Slave 模式，异步复制
每个 Master 配置一个或多个Slave，有多对 Master-Slave， HA（高可用集群）采取异步复制方法，主备有短暂消息延迟，毫秒级。

优点：
1、即使磁盘损坏，消息丢失很少，消息实时性不会被影响，因为 Master 宕机后，消费者仍然能够从 Slave 消费，此过程对应用透明。不需要人工干预。性能同多 Master 模式几乎一样。

缺点：
1、Master宕机，磁盘损坏时，因为主备有短暂消息延迟，未复制到slave消息会丢失。
2、现在master宕机后，备机不能自动切换为主机。只有master能够接收消息，若全部master宕机，将不能接收消息

4、多 Master 多 Slave 模式，同时双写
每个 Master 配置一个或多个Slave，有多对 Master-Slave， HA 采取同时双写方法，主备全部写成功，才返回成功。

优点：
数据和服务全部无单点， Master 宕机情冴下，消费者能够从slave消费、消息无延迟，服务可用性和数据可用性全部很高

缺点：
1、性能比异步复制模式略低，収送单个消息 RT（返回时间）会略高。
2、现在master宕机后，备机不能自动切换为主机。只有master能够接收消息，若全部master宕机，将不能接收消息

选择结论
因为我们需要确保消息中间件高可用性，消息不丢失、消息无延迟，所以我们选择“多Master多Slave模式，同时双写”模式。而且选择同时刷盘。
3、2、2 多Master多Slave模式

多master多slave模式网络结构图
关键组件有：Name Server、Broker、Producer、Consumer

1、Name Server是一个几乎无状态节点，可集群布署，节点之间无信息同时、统计Topic 路由信息。

2、Broker分为 Master和Slave，一个 Master 能够对应多个Slave，不过一个 Slave只能对应一个Master。

3、Producer和Name Server 集群中其中一个节点（随机选择）建立长连接，定时从Name Server获取Topic 路由信息，并向提供Topic服务Master 建立长连接，定时向 Master 发送心跳。Producer只能够向Master发送消息。Producer 完全无状态，可集群布署。

4、Consumer和Name Server 集群中其中一个节点（随机选择）建立长连接，定时从Name Server获取Topic路由信息，并和提供Topic服务Master、 Slave建立长连接，并定时向Master、Slave収送心跳。Consumer既能够从 Master 订阅消息，也能够从 Slave 订阅消息，订阅规则由Broker配置决定

3、3 集群搭建
linux 环境下布署rocketMq多master多slave模式、同时双写模式集群，
暂定为2个master，2个slave
3、3、1 安装条件
4台linux服务器、分为master-a、slave-a ; master-b、slave-b
服务器防火墙开启9876，10911
lokkit -p 9876:tcp -p 10911:tcp

服务器支持wget命令
服务器安装jdk，不低于使用rocketMq支持版本

3、3、2 安装步骤
4台linux服务器、分为master-a、slave-a ; master-b、slave-b
假设ip分别为：
master-a =10.1.236.1
slave–a =10.1.236.2
master-b =10.1.236.3
slave-b =10.1.236.4

3、3、2、1 master-a
1 从github下载RocketMQ安装包或源码自编译安装
wget
2 解压缩、并创建数据、日志目录
tar –xvf alibaba-rocketmq-3.2.2.tar.gz
3 配置环境变量:
系统变量：
Vi /etc/profile
或修改目前用户环境变量
比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq
export PATH=${PATH}:${ROCKETMQ_HOME}/bin
source 命令是环境变量生效
4 修改mq集群master-a 配置
修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties
不是强制必需使用这个文件，使用者能够自行定义
# brokerClusterName=DefaultCluster
brokerName=broker-a #归属master-slave组名字
brokerId=0 #0表示为master-slave组中为 master
namesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameserv
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
listenPort=10911 #Broker 对外服务监听端口
deleteWhen=04
fileReservedTime=120
mapedFileSizeCommitLog=
mapedFileSizeConsumeQueue=50000000
destroyMapedFileIntervalForcibly=10
redeleteHangedFileInterval=10
diskMaxUsedSpaceRatio=88
storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录
storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录
maxMessageSize=65536
flushCommitLogLeastPages=4
flushConsumeQueueLeastPages=2
flushCommitLogThoroughInterval=10000
flushConsumeQueueThoroughInterval=60000
checkTransactionMessageEnable=false
sendMessageThreadPoolNums=128
pullMessageThreadPoolNums=128
brokerRole=SYNC_MASTER #角色同时双写Master
flushDiskType=SYNC_FLUSH #同时刷盘
brokerIP1=10.1.236.1 #本机IP地址，多网卡易犯错，请手工指定

其它配置请参考《RocketMQ 开发指南》，最新版针对 v3.2.4

5 开启mq集群master-a
跳转到RocketMQbin目录下
>cd $ROCKETMQ_HOME/bin
>nohup sh mqnamesrv &
>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties &

3、3、2、2 slave-a
1从github下载RocketMQ安装包或源码自编译安装
wget
2 解压缩、并创建数据、日志目录
tar –xvf alibaba-rocketmq-3.2.2.tar.gz
3 配置环境变量
比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq
export PATH=${PATH}:${ROCKETMQ_HOME}/bin

source 命令是环境变量生效

4修改mq集群slave-a 配置
修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties
不是强制必需使用这个文件，使用者能够自行定义、只要确保配置文件内brokerName 正确即可

# brokerClusterName=DefaultCluster
brokerName=broker-a #归属master-slave组名字
brokerId=1 #1表示在master-slave组中为slave
namesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
listenPort=10911 #对外端口
deleteWhen=04
fileReservedTime=120
mapedFileSizeCommitLog=
mapedFileSizeConsumeQueue=50000000
destroyMapedFileIntervalForcibly=10
redeleteHangedFileInterval=10
diskMaxUsedSpaceRatio=88
storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放
storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放
maxMessageSize=65536
flushCommitLogLeastPages=4
flushConsumeQueueLeastPages=2
flushCommitLogThoroughInterval=10000
flushConsumeQueueThoroughInterval=60000
checkTransactionMessageEnable=false
sendMessageThreadPoolNums=128
pullMessageThreadPoolNums=128
brokerRole=SLAVE #角色 Slave
flushDiskType=SYNC_FLUSH # 同时刷盘
brokerIP1=10.1.236.2 #本机ip，多网卡，提议自定义

其它配置请参考《RocketMQ 开发指南》，最新版针对 v3.2.4

5 开启mq集群slave-a
跳转到RocketMQbin目录下
>cd $ROCKETMQ_HOME/bin
>nohup sh mqnamesrv &
>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties &

3、3、2、3 master-b
1 从github下载RocketMQ安装包或源码自编译安装
wget
2 解压缩、并创建数据、日志目录
tar –xvf alibaba-rocketmq-3.2.2.tar.gz
3 配置环境变量
比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq
export PATH=${PATH}:${ROCKETMQ_HOME}/bin

source 命令是环境变量生效
4 修改mq集群master-b配置
修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties
不是强制必需使用这个文件，使用者能够自行定义
# brokerClusterName=DefaultCluster
brokerName=broker-b #归属master-slave组名字
brokerId=0 #0表示为master-slave组中为 master
namesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameserv
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
listenPort=10911 #Broker 对外服务监听端口
deleteWhen=04
fileReservedTime=120
mapedFileSizeCommitLog=
mapedFileSizeConsumeQueue=50000000
destroyMapedFileIntervalForcibly=10
redeleteHangedFileInterval=10
diskMaxUsedSpaceRatio=88
storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录
storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录
maxMessageSize=65536
flushCommitLogLeastPages=4
flushConsumeQueueLeastPages=2
flushCommitLogThoroughInterval=10000
flushConsumeQueueThoroughInterval=60000
checkTransactionMessageEnable=false
sendMessageThreadPoolNums=128
pullMessageThreadPoolNums=128
brokerRole=SYNC_MASTER #角色同时双写Master
flushDiskType=SYNC_FLUSH #同时刷盘
brokerIP1=10.1.236.3 #本机IP地址，多网卡易犯错，请手工指定

其它配置请参考《RocketMQ 开发指南》，最新版针对 v3.2.4

5 开启mq集群master-b
跳转到RocketMQbin目录下
>cd $ROCKETMQ_HOME/bin
>nohup sh mqnamesrv &
>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties &

3、3、2、4 slave-b
1从github下载RocketMQ安装包或源码自编译安装
wget
2 解压缩、并创建数据、日志目录
tar –xvf alibaba-rocketmq-3.2.2.tar.gz
3 配置环境变量
比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq
export PATH=${PATH}:${ROCKETMQ_HOME}/bin

source 命令是环境变量生效

4修改mq集群slave-a 配置
修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties
不是强制必需使用这个文件，使用者能够自行定义、只要确保配置文件内brokerName 正确即可

# brokerClusterName=DefaultCluster
brokerName=broker-b #归属master-slave组名字
brokerId=1 #1表示在master-slave组中为slave
namesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
listenPort=10911 #对外端口
deleteWhen=04
fileReservedTime=120
mapedFileSizeCommitLog=
mapedFileSizeConsumeQueue=50000000
destroyMapedFileIntervalForcibly=10
redeleteHangedFileInterval=10
diskMaxUsedSpaceRatio=88
storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放
storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放
maxMessageSize=65536
flushCommitLogLeastPages=4
flushConsumeQueueLeastPages=2
flushCommitLogThoroughInterval=10000
flushConsumeQueueThoroughInterval=60000
checkTransactionMessageEnable=false
sendMessageThreadPoolNums=128
pullMessageThreadPoolNums=128
brokerRole=SLAVE #角色 Slave
flushDiskType=SYNC_FLUSH # 同时刷盘
brokerIP1=10.1.236.4 #本机ip，多网卡，提议自定义

其它配置请参考《RocketMQ 开发指南》，最新版针对 v3.2.4

5 开启mq集群slave-b
跳转到RocketMQbin目录下
>cd $ROCKETMQ_HOME/bin
>nohup sh mqnamesrv &
>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties &
4、MQ消息服务接口实现
4.1、步骤图

4.2、消息中间件接口规范
此服务接口以dubbo提供restful协议对外提供发送消息服务，并经过backurl回调消费端把消息推送给消费者，使用此服务能够经过http post请求方法，消费端要提供接收消息http协议post接口。
1、中间件接口
Rest路径：:xxxx/mq/send/sendMQ

入参(json)：{“backurl”:” :xxxx/xx”, //必填 消费端接收消息接口地址
“data”:{ //必填 要发送消息体
xxx:xxx,
xxx:xxx
……….
}
}
出参(json)：{"status":"success","msg":"发送成功"} //成功
{"status":"error","msg":"发送失败原因"} //失败

2、消费端接口

出参（String）：”success”//消费成功
“error” //消费失败