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

《RocketMq消息队列实施专项方案完整版.docx》由会员分享，可在线阅读，更多相关《RocketMq消息队列实施专项方案完整版.docx（16页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、消息队列实施方案1、背景异步解耦合、给前端系统提供最高效反应2、常见消息队列对比2、1 ActiveMqActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范 JMS Provider实现优点：Java语言支持集群模式缺点：性能在消息中间件中处于下游2、2 RabbitmqRabbitmq是基于AMQP使用erlang语言实现消息队列系统优点：1、完整消息队列系统，支持多个消息队列模式，包含竞争消费；2、支持集群模式，扩展集群容量和性能比较方便，集成了集群监控和管理；3、支持消息持久化；缺点：1、需要学习比较复杂接口和协议，比较花费时间；2、性能不是尤其理想大约在1wqps左右；3

2、、使用Erlang语言，语言基础；2、3 KafkaKafka 是LinkedIn 开发一个高性能、分布式消息公布订阅系统。优点：1、分布式集群能够透明扩展，增加新服务器进集群。2、高性能。单机写入TPS约在百万条/秒3、容错。数据全部会复制到几台服务器上。缺点：1、复杂性。Kafka需要zookeeper 集群支持，Topic通常需要人工来创建，布署和维护较通常消息队列成本更高定在日志传输、存在消息丢失肯能、消息乱序3、消息发送错误无重试2、4 RocketMQRockerMq 是阿里企业中间件团体参考Kafka思想，用Java语言实现消息传输系统优点：1、较高性能，单机写入TPS单实例约7

3、万条/秒2、容错，多个集群模式、能够处理容错问题3、消息重试发送4、次序消息能够严格实施缺点：1、消息反复、消费端需要做去重操作2、5 选择结论从项目业务和团体技术偏向考虑，我们应该需要一个数据安全性比较高，确保每个消息全部会被实施；有容错机制、支持集群模式高可用；性能不错，能够在毫秒级处理消息；支持次序消息消息中间件， RockerMq 能够满足这些要求。3、RockerMq介绍3、1 RockerMq 产品介绍参考阿里企业提供RocketMQ 开发指南，最新版针对 v3.2.43、2 RockerMq集群3、2、1 布署方法Rockermq共有四种布署方法，分别是：1、单个 Master一

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

5、秒级。优点：1、即使磁盘损坏，消息丢失很少，消息实时性不会被影响，因为 Master 宕机后，消费者仍然能够从 Slave 消费，此过程对应用透明。不需要人工干预。性能同多 Master 模式几乎一样。缺点： 1、Master宕机，磁盘损坏时，因为主备有短暂消息延迟，未复制到slave消息会丢失。2、现在master宕机后，备机不能自动切换为主机。只有master能够接收消息，若全部master宕机，将不能接收消息4、多 Master 多 Slave 模式，同时双写每个 Master 配置一个或多个Slave，有多对 Master-Slave， HA 采取同时双写方法，主备全部写成功，才返回成

6、功。优点：数据和服务全部无单点， Master 宕机情冴下，消费者能够从slave消费、消息无延迟，服务可用性和数据可用性全部很高缺点：1、性能比异步复制模式略低，収送单个消息 RT（返回时间）会略高。2、现在master宕机后，备机不能自动切换为主机。只有master能够接收消息，若全部master宕机，将不能接收消息选择结论因为我们需要确保消息中间件高可用性，消息不丢失、消息无延迟，所以我们选择“多Master多Slave模式，同时双写”模式。而且选择同时刷盘。3、2、2 多Master多Slave模式多master多slave模式网络结构图关键组件有：Name Server、Broker

7、、Producer、Consumer1、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、Con

8、sumer和Name Server 集群中其中一个节点（随机选择）建立长连接，定时从Name Server获取Topic路由信息，并和提供Topic服务Master、 Slave建立长连接，并定时向Master、Slave収送心跳。Consumer既能够从 Master 订阅消息，也能够从 Slave 订阅消息，订阅规则由Broker配置决定3、3 集群搭建linux 环境下布署rocketMq多master多slave模式、同时双写模式集群，暂定为2个master，2个slave3、3、1 安装条件4台linux服务器、分为master-a、slave-a ; master-b、slave-

9、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.1slavea =10.1.236.2master-b =10.1.236.3slave-b =10.1.236.43、3、2、1 master-a1 从github下载RocketMQ安装包或源码自编译安装wget 2 解压缩、并创建数据、日志目录 t

10、ar xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量:系统变量：Vi /etc/profile 或修改目前用户环境变量比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=$PATH:$ROCKETMQ_HOME/bin source 命令是环境变量生效4 修改mq集群master-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties不是强制必需使用这个文件，使用者能够自行定义# brokerClusterName=D

11、efaultClusterbrokerName=broker-a #归属master-slave组名字brokerId=0 #0表示为master-slave组中为 masternamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameservdefaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #Broker 对外服务监听端口deleteWhen=0

12、4fileReservedTime=120mapedFileSizeCommitLog=mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=10redeleteHangedFileInterval=10diskMaxUsedSpaceRatio=88storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录maxMessageSize=6

13、5536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SYNC_MASTER #角色同时双写MasterflushDiskType=SYNC_FLUSH #同时刷盘brokerI

14、P1=10.1.236.1 #本机IP地址，多网卡易犯错，请手工指定其它配置请参考RocketMQ 开发指南，最新版针对 v3.2.45 开启mq集群master-a跳转到RocketMQbin目录下cd $ROCKETMQ_HOME/binnohup sh mqnamesrv &nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties &3、3、2、2 slave-a1从github下载RocketMQ安装包或源码自编译安装wget 2 解压缩、并创建数据、日志目录tar xvf alibaba-rocke

15、tmq-3.2.2.tar.gz3 配置环境变量比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=$PATH:$ROCKETMQ_HOME/bin source 命令是环境变量生效4修改mq集群slave-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties不是强制必需使用这个文件，使用者能够自行定义、只要确保配置文件内brokerName 正确即可# brokerClusterName=DefaultClusterbrokerName=brok

16、er-a #归属master-slave组名字brokerId=1 #1表示在master-slave组中为slavenamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #对外端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=mapedFi

17、leSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=10redeleteHangedFileInterval=10diskMaxUsedSpaceRatio=88storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放maxMessageSize=65536flu

18、shCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SLAVE #角色 SlaveflushDiskType=SYNC_FLUSH # 同时刷盘brokerIP1=10.1.236.2 #本

19、机ip，多网卡，提议自定义其它配置请参考RocketMQ 开发指南，最新版针对 v3.2.45 开启mq集群slave-a跳转到RocketMQbin目录下cd $ROCKETMQ_HOME/binnohup sh mqnamesrv &nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties &3、3、2、3 master-b1 从github下载RocketMQ安装包或源码自编译安装wget 2 解压缩、并创建数据、日志目录 tar xvf alibaba-rocketmq-3.2.2.tar.gz3

20、 配置环境变量比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=$PATH:$ROCKETMQ_HOME/bin source 命令是环境变量生效4 修改mq集群master-b配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties不是强制必需使用这个文件，使用者能够自行定义# brokerClusterName=DefaultClusterbrokerName=broker-b #归属master-slave组名字brokerId=0 #0表示为mast

21、er-slave组中为 masternamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameservdefaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #Broker 对外服务监听端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=mapedFileSizeConsumeQueue=5

22、0000000destroyMapedFileIntervalForcibly=10redeleteHangedFileInterval=10diskMaxUsedSpaceRatio=88storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitL

23、ogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SYNC_MASTER #角色同时双写MasterflushDiskType=SYNC_FLUSH #同时刷盘brokerIP1=10.1.236.3 #本机IP地址，多网卡易犯错，请手工指定其它配置请参考RocketMQ 开发指南，最新版针对 v3.2.45 开启

24、mq集群master-b跳转到RocketMQbin目录下cd $ROCKETMQ_HOME/binnohup sh mqnamesrv &nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties &3、3、2、4 slave-b1从github下载RocketMQ安装包或源码自编译安装wget 2 解压缩、并创建数据、日志目录tar xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量比如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-r

25、ocketmqexport PATH=$PATH:$ROCKETMQ_HOME/bin source 命令是环境变量生效4修改mq集群slave-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties不是强制必需使用这个文件，使用者能够自行定义、只要确保配置文件内brokerName 正确即可# brokerClusterName=DefaultClusterbrokerName=broker-b #归属master-slave组名字brokerId=1 #1表示在master-slave组中为slavenamesrvAddr=

26、10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #对外端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=10redeleteH

27、angedFileInterval=10diskMaxUsedSpaceRatio=88storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoro

28、ughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SLAVE #角色 SlaveflushDiskType=SYNC_FLUSH # 同时刷盘brokerIP1=10.1.236.4 #本机ip，多网卡，提议自定义其它配置请参考RocketMQ 开发指南，最新版针对 v3.2.45 开启mq集群slave-b跳转到RocketM

29、Qbin目录下cd $ROCKETMQ_HOME/binnohup 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” /消费失败