RocketMQ安装部署

2023-12-18 11:21:33

RocketMQ 基础知识介绍

Apache RocketMQ是阿里开源的一款高性能、高吞吐量、队列模型的消息中间件的分布式消息中间件。

RocketMQ具有以下特点：
1）是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式特点。
2）Producer、Consumer、队列都可以分布式。
3）Producer向一些队列轮流发送消息，队列集合称为Topic，Consumer如果做广播消费，则一个consumer实例消费这个Topic对应的所有队列，如果做集群消费，则多个Consumer实例平均消费这个topic对应的队列集合。
4）支持严格的消息顺序；
5）提供丰富的消息拉取模式
6）高效的订阅者水平扩展能力
7）实时的消息订阅机制
8）亿级消息堆积能力
9）较少的依赖
10）支持Topic与Queue两种模式；
11）同时支持Push与Pull方式消费消息；

消息队列的应用场景
1）异步处理
将不是必须的业务逻辑，进行异步处理，比如注册之后短信、邮箱的发送

2）应用解耦
订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。
库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作。
假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。

3）流量削锋，也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛。
应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。
a）可以控制活动的人数；
b）可以缓解短时间内高流量压垮应用；
c）用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面；
d）秒杀业务根据消息队列中的请求信息，再做后续处理。

4）日志处理

5）消息通讯
消息通讯是指，消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等。

6）性能
RocketMQ单机也可以支持亿级的消息堆积能力。单机写入TPS单实例约7万条/秒，单机部署3个Broker，可以跑到最高12万条/秒，消息大小10个字节

RocketMQ的部署结构有以下特点：
1）Name Server 可集群部署，节点之间无任何信息同步。
2）Broker(消息中转角色，负责存储消息，转发消息) 部署相对复杂，Broker 分为Master 与Slave，一个Master 可以对应多个Slave，但是一个Slave 只能对应一个Master，Master 与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定 义，BrokerId为0 表示Master，非0 表示Slave。Master 也可以部署多个。
3）Producer 与Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从Name Server 取Topic 路由信息，并向提供Topic 服务的Master 建立长连接，且定时向Master 发送心跳。Producer 完全无状态，可集群部署。
4）Consumer 与Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从Name Server 取Topic 路由信息，并向提供Topic 服务的Master、Slave 建立长连接，且定时向Master、Slave 发送心跳。Consumer既可以从Master 订阅消息，也可以从Slave 订阅消息，订阅规则由Broker 配置决定。

Broker：消息中转角色，负责存储消息，转发消息

Broker集群有多种配置方式：
1）单Master
优点：除了配置简单没什么优点
缺点：不可靠，该机器重启或宕机，将导致整个服务不可用
2）多Master
优点：配置简单，性能最高
缺点：可能会有少量消息丢失（配置相关），单台机器重启或宕机期间，该机器下未被消费的消息在机器恢复前不可订阅，影响消息实时性
3）多Master多Slave，每个Master配一个Slave，有多对Master-Slave，集群采用异步复制方式，主备有短暂消息延迟，毫秒级
优点：性能同多Master几乎一样，实时性高，主备间切换对应用透明，不需人工干预
缺点：Master宕机或磁盘损坏时会有少量消息丢失
4）多Master多Slave，每个Master配一个Slave，有多对Master-Slave，集群采用同步双写方式，主备都写成功，向应用返回成功
优点：服务可用性与数据可用性非常高
缺点：性能比异步集群略低，当前版本主宕备不能自动切换为主

Master和Slave的配置文件参考conf目录下的配置文件
Master与Slave通过指定相同的brokerName参数来配对，Master的BrokerId必须是0，Slave的BrokerId必须是大于0的数。
一个Master下面可以挂载多个Slave，同一Master下的多个Slave通过指定不同的BrokerId来区分。

Rocketmq默认给出了三种建议配置模式: 2m-2s-async(主从异步); 2m-2s-sync(主从同步); 2m-noslave(仅master)

7）消息重试

MQ 消费者的消费逻辑失败时，可以通过设置返回状态达到消息重试的结果。
MQ 消息重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。

 

1.基础组件准备
JDK，alibaba-rocketmq-3.2.6.tar.gz，rocketmq-console.war

2. Hosts添加信息
192.168.200.24 rocketmq-nameserv1
192.168.200.24 rocketmq-master1
192.168.200.25 rocketmq-nameserv2
192.168.200.25 rocketmq-master2

ping rocketmq-nameserv1
ping rocketmq-master1
ping rocketmq-nameserv2
ping rocketmq-master2

3. 部署java环境jdk

#jdk的解压安装

[root@localhost ~]# tar xf jdk-8u60-linux-x64.tar.gz -C /usr/local/

[root@localhost ~]# ln -s /usr/local/jdk1.8.0_60 /usr/local/jdk

 

#配置java环境变量

[root@localhost ~]# sed -i.ori ‘$a export JAVA_HOME=/usr/local/jdk\nexport PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH\nexport CLASSPATH=.$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar‘ /etc/profile

[root@localhost ~]# tail -3 /etc/profile

export JAVA_HOME=/usr/local/jdk

export PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH

export CLASSPATH=.$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar

 

#让java环境变量立刻生效

[root@localhost ~]# source /etc/profile

 

#检查java环境安装情况

[root@localhost ~]# which java

/usr/local/jdk/bin/java

[root@localhost ~]# java -version       #出现以下信息表示部署成功

java version "1.8.0_60"

Java(TM) SE Runtime Environment (build 1.8.0_60-b27)

Java HotSpot(TM) 64-Bit Server VM (build 25.60-b23, mixed mode)

4上传解压【两台机器】
# 上传alibaba-rocketmq-3.2.6.tar.gz文件至
tar -zxvf alibaba-rocketmq-3.2.6.tar.gz -C /usr/local
cd /usr/local
mv alibaba-rocketmq alibaba-rocketmq-3.2.6
ln -s alibaba-rocketmq-3.2.6 rocketmq

ll /usr/local

5. 创建存储路径【两台机器】
cd /usr/local
mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index

或者
cd /usr/local
mkdir -p rocketmq/store/{commitlog,consumequeue,index}

6. RocketMQ主配置文件【两台机器】

  • 2m-2s-async ———– 异步复制
  • 2m-2s-sync ———— 同步双写
  • 2m-noslave ———— 多master模式


vim /usr/local/rocketmq/conf/2m-2s-asyn/broker-a.properties

vim /usr/local/rocketmq/conf/2m-2s-asyn/broker-b.properties

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a|broker-b
#0 表示 Master， >0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.200.24:9876;192.168.200.25:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#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
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false

#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

 

 RocketMQ从配置文件【两台机器】

vim /usr/local/rocketmq/conf/2m-2s-asyn/broker-b-s.properties
vim /usr/local/rocketmq/conf/2m-2s-asyn/broker-a-s.properties

#所属集群名字

borkerClusterName=rocketmq-cluster

#broker名字，注意此处不同的配置文件填写的不一样

brokerName=broker-a|broker-b

#0 表示Master, >0 表示Slave

brokerId=1

#nameServer地址，分号分割

namesrvAddr=192.168.200.24:9876;192.168.200.25:9876

#在发消息时，自动创建服务器不存在的topic,默认创建的队列数

defaultTopicQueueNums=4

#是否允许Broker自动创建Topic，建议线下开启，线上关闭

autoCreateTopicEnable=true

#是否允许Broker自动创建订阅组，建议线下开启，线上关闭

autoCreateSubscriptionGroup=true

#Broker 对外服务的监听端口

listenPort=10950

#删除文件时间点，默认凌晨4点

deleteWhen=04

#文件保留时间，默认48小时

fileReservedTime=120

#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

 

#Broker 的角色

#- ASYNC_MASTER 异步复制Master

#- SYNC_MASTER 同步双写Master

#- SLAVE

brokerRole=SLAVE

 

#刷盘方式

#- ASYNC_FLUSH  异步刷盘

#- SYNC_FLUSH 同步刷盘

flushDiskType=ASYNC_FLUSH

 

#checkTransactionMessageEnable=false

#发消息线程池数量

#sendMessageThreadPoolNums=128

#拉消息线程池数量

#pullMessageThreadPoolNums=128


-----------------------------------------------------------------------------------------------

 

7. 修改日志配置文件【两台机器】
mkdir -p /usr/local/rocketmq/logs
cd /usr/local/rocketmq/conf && sed -i ‘s#${user.home}#/usr/local/rocketmq#g‘ *.xml

#sed -i ‘s#${user.home}#/usr/local/rocketmq#g‘ *.xml

8. 修改启动脚本参数【两台机器】
vim /usr/local/rocketmq/bin/runbroker.sh

#============================================================
==================
# 开发环境JVM Configuration
#============================================================
==================
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m -XX:PermSize=128m -XX:MaxPermSize=320m"


vim /usr/local/rocketmq/bin/runserver.sh
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m -XX:PermSize=128m -XX:MaxPermSize=320m"

 

9. 启动 NameServer 【两台机器】
修改 NameServer 的端口（不修改忽略）
cd /usr/local/rocketmq/conf
vim namesrv.properties
listenPort=xxxx
启动 NameServer 【两台机器】
cd /usr/local/rocketmq/bin
nohup sh mqnamesrv -c/usr/local/rocketmq/conf/namesrv.properties >/dev/null 2>&1 &
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log


10. 启动 BrokerServer A【192.168.200.24】
cd /usr/local/rocketmq/bin

nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties >/dev/null 2>&1 &


tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log


netstat -ntlp
jps
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log

 

nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties >/dev/null 2>&1 &


11. 启动BrokerServer B【192.168.200.25】
cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties >/dev/null 2>&1 &
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log

 

netstat -ntlp
jps
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log

 

nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties >/dev/null 2>&1 &

停止服务
先停止BrokerServer，再停止NameServer
cd /usr/local/rocketmq/bin
sh mqshutdown broker
sh mqshutdown namesrv
 

 

12. rocketMq监控平台rocketmq-console部署

rocketmq有一个管理界面叫incubator-rocketmq-externals
百度网盘下载地址：https://pan.baidu.com/s/19PROiNivWBiOIV5NkOBBqA
提取密码：r1cf
   
也可以直接git在线下载
 cd /opt
 git clone https://github.com/apache/rocketmq-externals.git
   
做软链接
 ln -s /opt/rocketmq-externals /data/
 ll /data/rocketmq-externals
lrwxrwxrwx. 1 root root 23 May  9 14:10 /data/rocketmq-externals -> /opt/rocketmq-externals
   
修改配置文件
vim /data/rocketmq-externals/rocketmq-console/src/main/resources/application.properties
server.contextPath=
server.port=8080                              #默认访问端口是8080
#spring.application.index=true
spring.application.name=rocketmq-console
spring.http.encoding.charset=UTF-8
spring.http.encoding.enabled=true
spring.http.encoding.force=true
logging.config=classpath:logback.xml
#if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, you can set it in ops page.default localhost:9876
rocketmq.config.namesrvAddr=192.168.10.207:9876               #如果nameserver是多台集群机器，则后面就配置多个ip+port，即"rocketmq.config.namesrvAddr=ip1:port;ip2:port"
#if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
rocketmq.config.isVIPChannel=false                            #注意这个参数，是否设置为false或true取决于rocketmq的版本号
#rocketmq-console‘s data path:dashboard/monitor
rocketmq.config.dataPath=/tmp/rocketmq-console/data
#set it false if you don‘t want use dashboard.default true
rocketmq.config.enableDashBoardCollect=true
  
------------------------------------------------------------------------------------------
温馨提示：
由于我这里使用的rocketmq是3.2.6版本，低于3.5.8版本，所以上面的rocketmq.config.isVIPChannel应设置为false，默认为true！
Rocket如果开启了VIP通道，VIP通道端口为10911-2=10909。若Rocket服务器未启动端口10909，则报connect to <：10909> failed。
------------------------------------------------------------------------------------------
   
安装Maven的安装
cd /data/software/
 wget http://mirrors.hust.edu.cn/apache/maven/maven-3/3.3.9/binaries/apache-maven-3.3.9-bin.tar.gz
 tar -zvxf apache-maven-3.3.9-bin.tar.gz
 mv apache-maven-3.3.9 /usr/local/maven
vim /etc/profile
........
export MAVEN_HOME=/usr/local/maven
export PATH=$PATH:$MAVEN_HOME/bin
 
source /etc/profile
 
mvn --version
Apache Maven 3.3.9 (bb52d8502b132ec0a5a3f4c09453c07478323dc5; 2015-11-11T00:41:47+08:00)
Maven home: /usr/local/maven
Java version: 1.8.0_131, vendor: Oracle Corporation
Java home: /usr/java/jdk1.8.0_131/jre
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "2.6.32-696.el6.x86_64", arch: "amd64", family: "unix"
   
接着对rocketmq-console进行编译打包运行（mvn的编辑过程需要等待一段时间）
 
cd /data/rocketmq-externals/rocketmq-console/
 
mvn clean package -Dmaven.test.skip=true
   
编译成功后在target下找到文件：rocketmq-console-ng-1.0.0.jar
 
ls
doc  LICENSE  NOTICE  pom.xml  README.md  src  style  target
ls target/
checkstyle-cachefile    classes            maven-status                            rocketmq-console-ng-1.0.0-sources.jar
checkstyle-checker.xml  generated-sources  rocketmq-console-ng-1.0.0.jar
checkstyle-result.xml   maven-archiver     rocketmq-console-ng-1.0.0.jar.original
   
启动rocketmq-console，执行命令：
 
nohup java -jar target/rocketmq-console-ng-1.0.0.jar >/dev/null 2>&1 &
----------------------------------------------------------------------------------------------------
也可以直接跟NameServer的ip和端口进行启动
# java -jar rocketmq-console-ng-1.0.0.jar --server.port=8080 --rocketmq.config.namesrvAddr=192.168.10.207:9876
   
如果NameServer有多台集群机器的话，后面就多加
# java -jar rocketmq-console-ng-1.0.0.jar --server.port=8080 --rocketmq.config.namesrvAddr=192.168.10.207:9876;192.168.10.208:9876
 
这里需要注意两个参数：
--server.port=8080   指定console访问端口，默认的就是8080，也可以指定为其他端口
--rocketmq.config.namesrvAddr   指定nameserver的地址
----------------------------------------------------------------------------------------------------
   
rocketmq-console的默认端口是8080
lsof -i:8080
COMMAND  PID       USER   FD   TYPE   DEVICE SIZE/OFF NODE NAME
java    5574 confluence   81u  IPv6 24674767      0t0  TCP mq-console-nameserver:44888->mq-console-nameserver:webcache (CLOSE_WAIT)
java    5649       jira  157u  IPv6   879851      0t0  TCP *:webcache (LISTEN)
java    5649       jira  249u  IPv6  4166247      0t0  TCP mq-console-nameserver:46886->mq-console-nameserver:webcache (CLOSE_WAIT)
java    5649       jira  262u  IPv6 24669705      0t0  TCP mq-console-nameserver:43464->mq-console-nameserver:webcache (CLOSE_WAIT)
   
最后再浏览器里输入：http://192.168.10.207:8080，即可访问rockermq的web管理界面了


13.RocketMQ常用命令

需要切换到bin目录下，即：
cd /usr/local/rocketmq/bin
 
 
获取所有可用命令：
sh mqadmin
 
查看帮助：
# sh mqadmin <command> -h
查询Producer的网络连接情况：
# sh mqadmin producerConnection -n localhost:9876 -g <producer-group> -t <producer-topic>
查询Consumer的网络连接情况：
# sh mqadmin consumerConnection -n localhost:9876 -g <consumer-group>
查询Consumer的消费状态：
# sh mqadmin consumerProgress -n localhost:9876 -g <consumer-group>
 
查询消息是否发送成功
获取指定Topic：
# sh mqadmin topicList -n localhost:9876 | grep <topicName>
查看Topic状态：
# sh mqadmin topicStatus -n localhost:9876 -t <topicName>
根据offset获取消息：
# sh sh mqadmin queryMsgByOffset -n localhost:9876 -b <broker-name> -i <queueId> -o <offset> -t <topicName>
根据offsetMsgId查询消息：
# sh sh mqadmin queryMsgById -n localhost:9876 -i <offsetMsgId>
 
查询消息是否被消费成功
查询消息详情：
# sh mqadmin queryMsgById -i {MsgId} -n {NameServerAddr}
查看Consumer Group订阅了哪些TOPIC：
# sh mqadmin consumerProgress -g <ConsumerGroup> -n <NameServerAddr>
 
查询TOPIC被哪些Consumer Group订阅了
没有查询特定TOPIC订阅情况，只能查询所有后再过滤：
# sh mqadmin statsAll -n <NameServerAddr> | grep <TOPIC>
返回结果：#Topic #Consumer Group #InTPS #OutTPS #InMsg24Hour #OutMsg24Hour
 
关闭nameserver和所有的broker:
# sh mqshutdown namesrv
# sh mqshutdown broker
 
查看所有消费组group:
# sh mqadmin consumerProgress -n 192.168.23.159:9876
查看指定消费组（kevinGroupConsumer）下的所有topic数据堆积情况：
# sh mqadmin consumerProgress -n 192.168.23.159:9876 -g kevinGroupConsumer
查看所有topic :
# sh mqadmin topicList -n 192.168.23.159:9876
查看topic信息列表详情统计
# sh mqadmin topicstatus -n 192.168.23.159:9876 -t myTopicTest1
新增topic
# sh mqadmin updateTopic –n 10.45.47.168 –c DefaultCluster –t ZTEExample
删除topic
# sh mqadmin deleteTopic –n 10.45.47.168:9876 –c DefaultCluster –t ZTEExample

RocketMQ安装部署

上一篇：Vitual Studio 2019 密钥


下一篇：OpenStack实例配置虚拟IP