rocketmq4环境搭建（一）

àì夳堔傛蜴生んèń 2022-03-08 05:16 373阅读 0赞

### 文章目录 ###

*   *   *  消息队列好处
         *   *   *  1. 解耦:
                 *  2. 异步
                 *  3. 削峰（限流）
         *  TPS
         *  QPS
         *  消息队列缺点
         *   *   *  1. 系统可用性降低
                 *  2. 系统复杂度提高
                 *  3. 一致性问题
         *  RocketMQ
         *   *   *  1. 介绍
                 *  2. 优势
             *  专业术语
             *   *  Producer
                 *  Consumer
                 *  Push Consumer（我们一般用这种）
                 *  Producer Group
                 *  Consumer Group
                 *  Broker
                 *  广播消费
                 *  集群消费
                 *  RocketMQ消息是已文件记录形式持久化的。
                 *  队列集合称为Topic
             *  RocketMQ Producer集群物理部署结构
             *  Broker集群搭建
             *   *  多Master多Slave模式，异步复制
                 *  多Master多Slave模式，同步双写
             *  搭建单Master模式
             *   *  启动name server
                 *  启动broker（注意：发布部署时候需要指定配置文件发布）
                 *  停止name server 和 broker
                 *  启动name server，修改默认端口9876
             *  记录本地docker启动
             *  docker搭建rocketmq-console
             *   *  broker-a.properties属性说明
             *  使用工具模拟通信
             *  docker hub地址

### 消息队列好处 ###

##### 1. 解耦: #####

系统和系统之间，通过一个中间层的来进行协调，其他系统的调用就只要订阅这个MQ就可以进行业务处理。如果需要进行加入其他业务，只要订阅这个消息就可以了。 对于上游的业务层，根本不需要修改，他只和MQ对接，这样的话，和其他的业务系统进行了解耦。

##### 2. 异步 #####

对于我们以前的业务处理， 一些不重要的，不影响核心业务的处理也都是串行处理的，如果我们用MQ的话，对于一些不核心的业务可以订阅MQ，异步来出来，节省响应时间。

##### 3. 削峰（限流） #####

对于一些高峰期的大量用户进入的时候，我们系统可能支持不住，如果我们在前面用MQ把这些大量的请求放入队列中，等高峰期过去了， 然后我们在从队列中慢慢的处理堆积在MQ的请求，这样网站不至于崩溃了。

### TPS ###

TPS包括一条消息入和一条消息出，加上一次用户数据库访问。（业务TPS = CAPS × 每个呼叫平均TPS）

TPS是软件测试结果的测量单位。一个事务是指一个[客户机][Link 1]向服务器发送请求然后服务器做出反应的过程。客户机在发送请求时开始计时，收到服务器响应后结束计时，以此来计算使用的时间和完成的事务个数。

### QPS ###

QPS：每秒查询率QPS是对一个特定的查询服务器在规定时间内所处理流量多少的衡量标准，在因特网上，作为域名系统服务器的机器的性能经常用每秒查询率来衡量。对应fetches/sec，即每秒的响应请求数，也即是最大吞吐能力。

### 消息队列缺点 ###

##### 1. 系统可用性降低 #####

系统引入的外部依赖越多，越容易挂掉。本来你就是 A 系统调用 BCD 三个系统的接口就好了，人 ABCD 四个系统好好的，没啥问题，你偏加个 MQ 进来，万一 MQ 挂了咋整，MQ 一挂，整套系统崩溃的，你不就完了？如何保证消息队列的高可用。

##### 2. 系统复杂度提高 #####

硬生生加个 MQ 进来，你怎么\[保证消息没有重复消费\]？怎么\[处理消息丢失的情况\]？怎么保证消息传递的顺序性？头大头大，问题一大堆，痛苦不已。

##### 3. 一致性问题 #####

A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？你这数据就不一致了。

### RocketMQ ###

##### 1. 介绍 #####

它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。 RocketMQ目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

##### 2. 优势 #####

a. 面向分布式系统的MQ

b. 强调集群无单点，可扩展，任意一点高可用，水平扩展

c. 海量消息堆积能力，消息堆积后，写入低延迟

d. 支持上万队列

e. 消息失败重试机制

f. 消息可查询

#### 专业术语 ####

##### Producer #####

消息生产者，负责产生消息，一般由业务系统负责产生消息。

##### Consumer #####

消息消费者，负责消费消息，一般是后台系统负责异步消费。

##### Push Consumer（我们一般用这种） #####

Consumer 的一种，应用通常吐 Consumer 对象注册一个 Listener 接口，一旦收到消息，Consumer 对象立 刻回调 Listener 接口方法。

##### Producer Group #####

一类 Producer 的集合名称，返类 Producer 通常収送一类消息，丏収送逡辑一致。

##### Consumer Group #####

一类 Consumer 的集合名称，返类 Consumer 通常消费一类消息，丏消费逡辑一致。

##### Broker #####

消息中转角色，负责存储消息，转収消息，一般也称为 Server。在 JMS 规范中称为 Provider。

##### 广播消费 #####

一条消息被多个 Consumer 消费，即使返些 Consumer 属亍同一个 Consumer Group，消息也会被 Consumer Group 中的每个 Consumer 都消费一次，广播消费中的 Consumer Group 概念可以讣为在消息划分方面无意 丿。

##### 集群消费 #####

一个 Consumer Group 中的 Consumer 实例平均分摊消费消息。例如某个 Topic 有 9 条消息，其中一个 Consumer Group 有 3 个实例（可能是 3 个迕程，戒者 3 台机器），那举每个实例只消费其中的 3 条消息。

##### RocketMQ消息是已文件记录形式持久化的。 #####

##### 队列集合称为Topic #####

#### RocketMQ Producer集群物理部署结构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpdXlpbnRocmVl_size_16_color_FFFFFF_t_70]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpdXlpbnRocmVl_size_16_color_FFFFFF_t_70 1]

Topic可以理解为在rocketMq体系当中作为一个逻辑消息组织形式，一般情况下一类业务消息会申请一个topic来实现业务之间隔离。

Topic是一个逻辑上的概念，实际上在每个broker上以queue的形式保存，也就是说每个topic在broker上会划分成几个逻辑队列，每个逻辑队列保存一部分消息数据，但是保存的消息数据实际上不是真正的消息数据，而是指向commit log的消息索引。

#### Broker集群搭建 ####

推荐的几种 Broker 集群部署方式，返里的 Slave 不可写，但可读，类似亍 Mysql 主备方式。下面是常用的推荐模式

##### 多Master多Slave模式，异步复制 #####

主从节点数据同步方式为：异步复制 ，相当于，发送消息过来了，先到master，然后马上返回给客户端，同时，这个时候起一条线程，把数据复制到slave中。 这样，这种模式就是有可能在复制到slave的时候，消息丢失了。

##### 多Master多Slave模式，同步双写 #####

主从节点数据同步方式为：异步复制 ，相当于，发送消息过来了，先到master，然后数据复制到slave中，最后在返回给producer，这样的话，消息不会丢失，但是性能会有减少。

以上 Broker 不 Slave 配对是通过挃定相同的 brokerName 参数来配对，Master 的 BrokerId 必须是 0，Slave 的 BrokerId 必须是大亍 0 的数。另外一个 Master 下面可以挂载多个 Slave，同一 Master 下的多个 Slave 通过挃 定丌同的 BrokerId 来区分。

#### 搭建单Master模式 ####

查看官网， 下载提供的zip包

##### 启动name server #####

> nohup sh bin/mqnamesrv &
     > tail -f ~/logs/rocketmqlogs/namesrv.log
       The Name Server boot success...

##### 启动broker（注意：发布部署时候需要指定配置文件发布） #####

> nohup sh bin/mqbroker -n 192.168.70.102:9876 &
     > tail -f ~/logs/rocketmqlogs/broker.log 
       The broker[%s, 172.30.30.233:10911] boot success...
       
       注意：发布部署时候需要指定配置文件发布
       nohup sh bin/mqbroker -c  conf/2m-noslave/broker-a.properties -n 192.168.70.102:9876 &

需要注意的是默认的name server 启动是内存4G， broker是默认8G，我们需要修改runserver.sh和runbroker.sh中的JAVA\_OPT，修改小一点如：JAVA\_OPT="$\{JAVA\_OPT\} -server -Xms256m -Xmx256m -Xmn128m"，然后就可以启动成功了， 不然启动后电脑太卡。

##### 停止name server 和 broker #####

> sh bin/mqshutdown broker
    The mqbroker(36695) is running...
    Send shutdown request to mqbroker(36695) OK
    
    > sh bin/mqshutdown namesrv
    The mqnamesrv(36664) is running...
    Send shutdown request to mqnamesrv(36664) OK

##### 启动name server，修改默认端口9876 #####

在/opt/rocketmq/rocketmq-all-4.4.0-bin-release/conf中创建namesrv.properties文件， 加入listenPort=8876，然后下面直接-c 指定这个配置文件启动，启动后，server的端口就是你设置的，如：8876

nohup sh  bin/mqnamesrv  -c  conf/namesrv.properties > ~/logs/rocketmqlogs/namesrv.log 2>&1
    
    nohup sh  bin/mqnamesrv  -c  conf/namesrv.properties &

#### 记录本地docker启动 ####

docker run --privileged -itd -p 8022:22 -p 9876:9876 -p 10911:10911 -p 10909:10909  rmq_master /usr/sbin/init

#### docker搭建rocketmq-console ####

docker pull styletang/rocketmq-console-ng
    
    docker run -e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.70.102:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8086:8080 -t styletang/rocketmq-console-ng

##### broker-a.properties属性说明 #####

# 所属集群名字
    brokerClusterName=rocketmq-cluster
    # 连接外网地址， 如果不设置，会默认用你虚拟机或者本地机子ip 
    brokerIP1=192.168.70.102
    # ~此处变化 broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-a
    # 0 表示Master, > 0 表示slave
    brokerId=0
    # `此处变化（此处nameserver跟host配置相匹配，9876为默认rk服务默认端口）nameServer 地址，分号分割
    namesrvAddr=192.168.70.102:9876
    # Broker 对外服务的监听端口
    listenPort=10911
    # 删除文件时间点，默认是凌晨4点
    deleteWhen=04
    # 文件保留时间，默认48小时
    fileReservedTime=120
    # Broker 的角色
    # - ASYNC_MASTER 异步复制Master
    # - SYNC_MASTER 同步双写Master
    # - SLAVE
    brokerRole=ASYNC_MASTER
    # 刷盘方式
    # - ASYNC_FLUSH 异步刷盘
    # - SYNC_FLUSH 同步刷盘
    flushDiskType=ASYNC_FLUSH
    # #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭 (这个在模拟tools工具发送消息时候要打开，不然阻塞)
    autoCreateTopicEnable=true
    
    
    #######################下面属性后面再进行配置查看效果，先记录#####################
    
    
    
    #在发送消息时,自动创建服务器不存在的topic,默认创建的队列数
    defaultTopicQueueNums=4
    #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭
    autoCreateTopicEnable=true
    #是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭
    autoCreateSubscriptionGroup=true
    #commitLog每个文件的大小默认1G
    mapedFileSizeCommitLog=1073741824
    #ConsumeQueue每个文件默认存30W条,根据业务情况调整
    mapedFileSizeConsumeQueue=300000
    #destroyMapedFileIntervalForcibly=120000
    #redeleteHangedFileInterval=120000
    #检测物理文件磁盘空间
    diskMaxUsedSpaceRatio=88
    
    #这里是我的 日志配置
    #存储路径
    storePathRootDir=/usr/local/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/abort
    
    #限制的消息大小  
    maxMessageSize=65536
    #flushCommitLogLeastPages=4
    #flushConsumeQueueLeastPages=2
    #flushCommitLogThoroughInterval=10000
    #flushConsumeQueueThoroughInterval=60000
    #checkTransactionMessageEnable=false
    #发消息线程池数量
    #sendMessageThreadPoolNums=128
    #拉消息线程池数量
    #pullMessageThreadPoolNums=128

#### 使用工具模拟通信 ####

#生产端
    export NAMESRV_ADDR=192.168.70.102:9876
    sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
    
    #消费端
    sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

#### docker hub地址 ####

[Link 1]: https://www.baidu.com/s?wd=%E5%AE%A2%E6%88%B7%E6%9C%BA&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpdXlpbnRocmVl_size_16_color_FFFFFF_t_70]: /images/20220308/d0b5313ff8ca43799813e11bf355f93c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpdXlpbnRocmVl_size_16_color_FFFFFF_t_70 1]: /images/20220308/8670b2d1e3a74416a75f4bc821ea5600.png