RocketMq消息队列实施方案

《RocketMq消息队列实施方案》由会员分享，可在线阅读，更多相关《RocketMq消息队列实施方案（13页珍藏版）》请在装配图网上搜索。

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

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

3、 Kafka思想，用 Java 语言实现的消息传输系统优点：1、较高性能，单机写入 TPS 单实例约 7 万条 /秒2、容错，多种集群模式、可以解决容错问题3、消息重试发送4、顺序消息可以严格执行 缺点：1、消息重复、消费端需要做去重操作2、 5 选用结论从项目业务与团队技术偏向考虑，我们应该需要一种数据安全性比较高， 保证每个消息都会被执行；有容错机制、支持集群模式高可用；性能不错，可以在毫秒级处理消息；支持顺序消息的消息中间件，RockerMq可以满足这些要求。3、RockerMq 简介3、 1 RockerMq 产品介绍参考阿里公司提供的 RocketMQ 开发指南，最新版针对 v3.2

4、.43、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 模式，异

5、步复制每个 Master 配置一个或多个 Slave ，有多对 Master-Slave ， HA（高可用集群） 采用异步 复制方式，主备有短暂消息延迟，毫秒级。优点：1、即使磁盘损坏，消息丢失的非常少，消息实时性不会被影响，因为 Master 宕机后，消 费者仍然可以从 Slave 消费，此过程对应用透明。不需要人工干预。性能同多 Master 模 式几乎一样。缺点：1、 Master 宕机，磁盘损坏时，因为主备有短暂消息延迟，未复制到 slave 的消息会丢失。2、目前 master 宕机后，备机不能自动切换为主机。只有 master 可以接收消息，若所有 master 宕机，将不能接收消

6、息4 、多 Master 多 Slave 模式，同步双写每个 Master 配置一个或多个 Slave ，有多对 Master-Slave ， HA 采用同步双写方式，主 备都写成功，才返回成功。优点：数据与服务都无单点， Master 宕机情冴下，消费者可以从 slave 消费、消息无延迟，服务可用性与数据可用性都非常高缺点：1、性能比异步复制模式略低，収送单个消息的RT （返回时间）会略高。2、目前 master 宕机后，备机不能自动切换为主机。只有 master 可以接收消息，若所有 master 宕机，将不能接收消息选用结论由于我们需要保证消息中间件的高可用性，消息不丢失、消息无延迟，

7、所以我们选择“多Master 多 Slave 模式，同步双写”模式。并且选择同步刷盘。3、2、2 多 Master多 Slave模式多 master 多 slave 模式网络结构图主要组件有：Name Server 、 Broker 、 Producer 、Consumer1、 Name Server 是一个几乎无状态节点，可集群部署，节点之间无信息同步、记录 Topic 路由信息。2、 Broker 分为 Master 和 Slave ，一个 Master 可以对应多个 Slave ，但是一个 Slave 只 能对应一个 Master 。3、Producer 与 Name Server 集群

8、中的其中一个节点 （随机选择） 建立长连接， 定期从 Name Server 获取 Topic 路由信息，并向提供 Topic 服务的 Master 建立长连接，定时向 Master 发送心跳。 Producer 只可以向 Master 发送消息。 Producer 完全无状态，可集群部署。4、Consumer 与 Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从 Name Server 获取 Topic 路由信息，并与提供 Topic 服务的 Master 、 Slave 建立长连接， 并定时向 Master 、Slave 収送心跳。 Consumer 既可以从 M

9、aster 订阅消息， 也可以从 Slave 订阅消息，订阅规则由 Broker 配置决定3、 3 集群搭建linux 环境下部署 rocketMq 多 master 多 slave 模式、同步双写模式集群， 暂定为 2 个 master ， 2 个 slave3、3、 1 安装条件4 台 linux 服务器、分为 master-a 、 slave-a ; master-b 、 slave-b 服务器防火墙开启 9876 ，10911lokkit -p 9876:tcp -p 10911:tcp服务器支持 wget 命令服务器安装 jdk ，不低于使用的 rocketMq 的支持版本3、3、

10、2 安装步骤4 台 linux 服务器、分为 master-a 、 slave-a ; master-b 、 slave-b 假设 ip 分别为：master-a =10.1.236.1slave a =10.1.236.2master-b =10.1.236.3slave-b =10.1.236.43、 3、2、1 master-a1 从 github 下载 RocketMQ 安装包或源码自编译安装 wget2 解压缩、并创建数据、日志目录 tar xvfa libaba-rocketmq-3.2.2.tar.gz3 配置环境变量 : 系统变量：Vi /etc/profile 或者修改当前用

11、户的环境变量例如： 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=DefaultClusterbrokerName=broker-a # 归属 master-slave 组的名字 brokerId

12、=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=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=true listenPort=10911 #Broker 对外服务的监听端口 deleteWhen=04=120 maped=1073741824 maped= destroyMaped=120000 r

13、edeleteHanged=120000 diskMaxUsedSpaceRatio=88 storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data# 数据目录storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs # 日志目录 maxMessageSize=65536flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueTho

14、roughInterval=60000 checkTransactionMessageEnable=false sendMessageThreadPoolNums=128 pullMessageThreadPoolNums=128 brokerRole=SYNC_MASTER #角色同步双写 Master flushDiskType=SYNC_FLUSH #同步刷盘 brokerIP1=10.1.236.1 #本机 IP 地址，多网卡易出错，请手工指定 其他配置请参考 RocketMQ 开发指南 ，最新版针对 v3.2.45 启动 mq 集群的 master-a 跳转到 RocketMQ 的

15、bin 目录下 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-a1 从 github 下载 RocketMQ 安装包或源码自编译安装wget2 解压缩、并创建数据、日志目录tar xvfa libaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如： export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq export

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

17、vAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 defaultTopicQueueNums=4autoCreateTopicEnable=true autoCreateSubscriptionGroup=true listenPort=10911 #对外端口 deleteWhen=04=120 maped=1073741824 maped= destroyMaped=120000 redeleteHanged=120000 diskMaxUsedSpaceRatio=88 storePathRootDi

18、r=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data#数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs # 日志存放 maxMessageSize=65536flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueThoroughInterval=6000

19、0 checkTransactionMessageEnable=false sendMessageThreadPoolNums=128 pullMessageThreadPoolNums=128 brokerRole=SLAVE #角色 Slave flushDiskType=SYNC_FLUSH # 同步刷盘 brokerIP1=10.1.236.2 #本机 ip，多网卡，建议自定义其他配置请参考 RocketMQ 开发指南 ，最新版针对 v3.2.45 启动 mq 集群的 slave-a 跳转到 RocketMQ的 bin 目录下 cd $ROCKETMQ_HOME/bin nohup s

20、h mqnamesrv &nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties &3、 3、2、3 master-b1 从 github 下载 RocketMQ 安装包或源码自编译安装wget2 解压缩、并创建数据、日志目录tar xvfa libaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如： export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmq export PATH=$PATH:$ROCKETMQ_HOME/binsour

21、ce 命令是环境变量生效4 修改 mq 集群的 master-b 配置修改文件 $ROCKETMQ HOME/conf/2m-2s-sync/broker-b.properties不是强制必须使用这个文件，使用者可以自行定义# brokerClusterName=DefaultClusterbrokerName=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.

22、4:9876 #nameserv defaultTopicQueueNums=4autoCreateTopicEnable=true autoCreateSubscriptionGroup=true listenPort=10911 #Broker 对外服务的监听端口 deleteWhen=04=120maped=1073741824maped=destroyMaped=120000 redeleteHanged=120000 diskMaxUsedSpaceRatio=88 storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data# 数据目录s

23、torePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs # 日志目录 maxMessageSize=65536flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueThoroughInterval=60000 checkTransactionMessageEnable=false sendMessageThreadPoolNums=128 pullMessageThreadPo

24、olNums=128brokerRole=SYNC_MASTER #角色同步双写 MasterflushDiskType=SYNC_FLUSH #同步刷盘brokerIP1=10.1.236.3 #本机 IP 地址，多网卡易出错，请手工指定其他配置请参考 RocketMQ 开发指南 ，最新版针对 v3.2.45 启动 mq 集群的 master-b 跳转到 RocketMQ的 bin 目录下 cd $ROCKETMQ_HOME/bin nohup sh mqnamesrv & nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-

25、b.properties &3、 3、2、4 slave-b1 从 github 下载 RocketMQ 安装包或源码自编译安装wget2 解压缩、并创建数据、日志目录tar xvfa libaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如： export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=$PATH:$ROCKETMQ_HOME/binsource 命令是环境变量生效4 修改 mq 集群的 slave-a 配置修改文件 $ROCKETMQ_HOME/conf/2m-2s-sync/broker

26、-b-s.properties 不是强制必须使用这个文件， 使用者可以自行定义、 只要保证配置文件内的 brokerName 确即可# brokerClusterName=DefaultClusterbrokerName=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=4autoCreateTopicEn

27、able=true autoCreateSubscriptionGroup=true listenPort=10911 #对外端口 deleteWhen=04=120 maped=1073741824 maped= destroyMaped=120000 redeleteHanged=120000diskMaxUsedSpaceRatio=88 storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data#数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/R

28、ocketMQ/alibaba-rocketmq/logs # 日志存放 maxMessageSize=65536flushCommitLogLeastPages=4 flushConsumeQueueLeastPages=2 flushCommitLogThoroughInterval=10000 flushConsumeQueueThoroughInterval=60000 checkTransactionMessageEnable=false sendMessageThreadPoolNums=128 pullMessageThreadPoolNums=128 brokerRole=SL

29、AVE #角色 Slave flushDiskType=SYNC_FLUSH # 同步刷盘 brokerIP1=10.1.236.4 #本机 ip，多网卡，建议自定义其他配置请参考 RocketMQ 开发指南 ，最新版针对 v3.2.45 启动 mq 集群的 slave-b 跳转到 RocketMQ 的 bin 目录下 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、流程图N

30、ameserv消息队列消息生产者消息队列消息消费者10缓存（ redis ）邮件系统1 消息队列启动时，将机器路由信息记录到 NameServ2 业务系统将发送消息请求到消息队列消息生产者3 消息生产者到 Namserv 获取消息队列服务端的路由列表4 获取消息队列服务端路由列表返回5 消息生产者中路由列表中获取一台消息队列服务器，发送消息到队列6 消息队列服务端推送消息到消息队列消费者（或消费者主动拉取）7消息消费者发送请求到业务系统，完成消息对应的业务逻辑（要求，最快返回成功）8消息消费者将消息处理结果通知服务端，成功或失败（失败时有三次重发机制）9同一消息消费三次均失败时，记录到 red

31、is10同一消息消费三次均失败时，发送消息给对应运营人员4.2、消息中间件接口规范此服务接口以 dubbo 提供的 restful 协议对外提供发送消息服务， 并通过 backurl 回调消费 端把消息推送给消费者，使用此服务可以通过 http post 请求的方式，消费端要提供接受消 息的 http 协议的 post 接口。1、中间件接口 Rest 路径：入参（json）：“backurl ”:”/ 必”填, 消费端接收消息的接口地址“data ”:/ 必填 要发送的消息体xxx:xxx, xxx:xxx出参 (json) ： status:success,msg:发送成功 / 成功status:error,msg: 发送失败原因 / 失败2、消费端接口出参（ String ）： ” success ”消/费成功 “ error ”/ 消费失败