zxpnet网站
shollin
2021-08-05
目录

Untitled

# rocketmq4.9安装

# 下载并解压

wget https://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.9.0/rocketmq-all-4.9.0-bin-release.zip
unzip rocketmq-all-4.9.0-bin-release.zip
1
2

# 检查内存设置

RocketMQ默认需要的JVM内存时比较大的。NameServer的默认启动参数（由bin目录下的runserver.sh文件指定），默认4g，改为256m 128m

Broker的默认启动参数（由bin目录下的runbroker.sh文件指定）,默认8g。 更改成512m 256m即可

# 启动nameserver和broker

可以将rocketmq的解压目录配置到/etc/profile里面，这样可以便捷启动

nohup sh bin/mqnamesrv &  #后台进程启动nameserver
nohup sh bin/mqbroker -n localhost:9876 &

# 查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log
tail -f ~/logs/rocketmqlogs/broker.log 

jps  #查看java程序
显示有 BrokerStartup、NamesrvStartup即启动正常

sh bin/mqshutdown namesrv  #关闭nameserver
sh bin/mqshutdown broker  #关闭broker
1
2
3
4
5
6
7
8
9
10
11
12

# rocketmq集群

preview

# 集群特点

  • NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

  • Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

  • Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

  • Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

img

# 搭建步骤

  • 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
  • Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
  • 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
  • Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
  • Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

# 准备两台服务器

服务器 nameserver broker 端口
a号服务器192.168.56.160 nameserver1 broker-master1,broker-slave2 nameserver:9876 broker-master:10911 broker-slave:11011
b号服务器192.168.56.161 nameserver2 broker-master2,broker-slave1 nameserver:9876 broker-master:10911 broker-slave:11011

1、设置/etc/hosts、 /etc/profile

2、创建mq保存目录

3、设置防火墙、

4、设置a服务器的broker-a.properties和broker-b-s.properties

5、设置b服务器的broker-b.properties和broker-a-s.properties

192.168.56.160 rocketmq-nameserver1 rocketmq-broker-master01 rocketmq-broker-slave02 host160
192.168.56.161 rocketmq-nameserver2 rocketmq-broker-master02 rocketmq-broker-slave01 host161

ping rocketmq-broker-master01 #看能否ping通
1
2
3
4
# 开放name server默认端口
firewall-cmd --remove-port=9876/tcp --permanent
# 开放master默认端口
firewall-cmd --remove-port=10911/tcp --permanent
# 开放slave默认端口 (当前集群模式可不开启)
firewall-cmd --remove-port=11011/tcp --permanent 
# 重启防火墙
firewall-cmd --reload

# 关闭防火墙
systemctl stop firewalld.service 
# 查看防火墙的状态
firewall-cmd --state 
# 禁止firewall开机启动
systemctl disable firewalld.service
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# a服务器
mkdir -p /mnt/data/rocketmq/store/broker-a/{commitlog,consumequeue,index}
mkdir -p /mnt/data/rocketmq/store/broker-b-s/{commitlog,consumequeue,index}

#b服务器
mkdir -p /mnt/data/rocketmq/store/broker-b/{commitlog,consumequeue,index}
mkdir -p /mnt/data/rocketmq/store/broker-a-s/{commitlog,consumequeue,index}
1
2
3
4
5
6
7

broker-a.properties broker-master配置,brokerName需要一样，brokerId=0为主，大于0为从

#暴露的外网IP，主节点要配置brokerIP2,与brokerIP1相同
brokerIP1=1号服务器公网IP
brokerIP2=1号服务器公网IP
#所属集群名字
brokerClusterName=zxp-rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2: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=/mnt/data/rocketmq/store/broker-a
#commitLog 存储路径
storePathCommitLog=/mnt/data/rocketmq/store/broker-a/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/mnt/data/rocketmq/store/broker-a/consumequeue
#消息索引存储路径
storePathIndex=/mnt/data/rocketmq/store/broker-a/index
#checkpoint 文件存储路径
storeCheckpoint=/mnt/data/rocketmq/store/broker-a/checkpoint
#abort 文件存储路径
abortFile=/mnt/data/rocketmq/store/broker-a/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

broker-b-s.properties

#暴露的外网IP，从节点不需要配置brokerIP2，它会根据brokerName找到主节点的brokerIP2
brokerIP1=1号服务器公网IP
#所属集群名字
brokerClusterName=zxp-rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/mnt/data/rocketmq/store/broker-b-s
#commitLog 存储路径
storePathCommitLog=/mnt/data/rocketmq/store/broker-b-s/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/mnt/data/rocketmq/store/broker-b-s/consumequeue
#消息索引存储路径
storePathIndex=/mnt/data/rocketmq/store/broker-b-s/index
#checkpoint 文件存储路径
storeCheckpoint=/mnt/data/rocketmq/store/broker-b-s/checkpoint
#abort 文件存储路径
abortFile=/mnt/data/rocketmq/store/broker-b-s/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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

brokerIP1 当前broker监听的IP brokerIP2 配置在主节点上，当存在broker主从时，在broker主节点上配置了brokerIP2的话,broker从节点会连接主节点配置的brokerIP2来同步。

默认不配置brokerIP1和brokerIP2时，都会根据当前网卡选择一个IP使用，当你的机器有多块网卡时，很有可能会有问题。

image-20210807172520539

  • broker-a.properties和broker-b-s.properties两个节点的持久化目录不可配置成相同的，并且不能手动创建持久化目录
  • 多网卡的问题,可以指定brokerIP1,来暴露外网ip

# 启动服务器

# 先启动两台服务器的nameserver

启动前检查jvm设置，nameserver为无状态，因此需要先启动，broker启动时，会来nameserver注册

cd /mnt/app/rock	
nohup bin/mqnamesrv &
1
2
# 再启动broker
# a服务器
nohup mqbroker -c /mnt/app/rocketmq-all-4.9.0-bin-release/conf/2m-2s-sync/broker-a.properties &
nohup bin/mqbroker -c /mnt/app/rocketmq-all-4.9.0-bin-release/conf/2m-2s-sync/broker-b-s.properties &

# b服务器
nohup mqbroker -c /mnt/app/rocketmq-all-4.9.0-bin-release/conf/2m-2s-sync/broker-b.properties &
nohup mqbroker -c /mnt/app/rocketmq-all-4.9.0-bin-release/conf/2m-2s-sync/broker-a-s.properties &

jps 查看是否启动成功
1
2
3
4
5
6
7
8
9

# 查看启动日志

# 查看nameServer日志
tail -500f ~/logs/rocketmqlogs/namesrv.log
# 查看broker日志
tail -500f ~/logs/rocketmqlogs/broker.log
1
2
3
4

# broker路由注册、剔除机制

路由注册、剔除机制说明如下：

  • 集群中所有Broker每隔30s向集群中所有的NameServer发送心跳包，注册Topic路由信息。
  • NameServer在收到Broker端的心跳包时首先会更新路由表，并记录收到心跳包的时间。
  • NameServer会启动一个定时任务每10s会扫描Broker，如果Nameserver连续120s未收到Broker的心跳包，会判定Broker已下线，将从路由表中将该Broker移除。如果Nameserver与Broker端的长连接断开，NameServer会立即感知Broker下线并从路由表中将该Broker移除。
  • 消息客户端(消息发送者、消息消费者)在任意时刻只会和其中一台NameServer建立连接，并每隔30s向NameServer查询路由信息，如果查询到结果会更新发送者的本地路由信息。

# mq监控可视化平台搭建

# 1、下载源码包

git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true
1
2
3

# 2、修改配置信息

server.port=18080
# cluster  /etc/hosts  rocketmq-nameserver1 split by ';'
rocketmq.config.namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
rocketmq.config.dataPath=/mnt/data/rocketmq/tmp/rocketmq-console/data
1
2
3
4

这里修改了默认端口8080，集群的nameserver用hosts别名来代替，中间用英文“;”隔开

# 3、打jar包

用idea打开，直接运行package即可，打包时，maven-checkstyle-plugin会进行代码检查，参考IntelliJ IDEA自动导入包去除星号（import xxx.*） (opens new window)

在pom中设置属性，忽略测试

<maven.test.skip>true</maven.test.skip>
1

# 4、将jar包上传到服务器

hostname  #查看主机名，如果是英文名，最好设置/etc/hosts:  192.168.56.160 zxp-host160

nohup java -jar rocketmq-console-ng-2.0.0.jar &
1
2
3

访问即可：http://192.168.56.160:18080/

# 消息发送

@Slf4j
public class SyncProducer {

    public static void main(String[] args) throws Exception {
        //1、 实例化消息生产者Producer
        DefaultMQProducer mqProducer = new DefaultMQProducer("GP-syncProducer");

        //2、 设置NameServer的地址，注意broker-a.properties要暴露公网ip
        mqProducer.setNamesrvAddr("rocketmq-nameserver1:9876;rocketmq-nameserver2:9876");

        //3、启动Producer实例
        mqProducer.start();
        for (int i = 0; i < 10; i++) {
            // 4、创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("topic1","tag1",("msg"+i).getBytes());

            // 5、发送消息到一个Broker
            SendResult result = mqProducer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", result);
            TimeUnit.SECONDS.sleep(1);
        }

        // 6、如果不再发送消息，关闭Producer实例。
        mqProducer.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

参考文章：

高可用RocketMQ集群搭建 - 知乎 (zhihu.com) (opens new window)

RocketMQ〖一〗介绍及在服务器上分布式搭建多Master多Slave模式RocketMQ_kingtok的博客-CSDN博客 (opens new window)

参考视频

黑马程序员Java教程RocketMQ系统精讲，电商分布式消息中间件，硬核揭秘双十一_哔哩哔哩_bilibili (opens new window)

【No1001】H马全面解剖RocketMQ和项目实战（完整资料）-百度网盘下载-Java学习者论坛 (javaxxz.com) (opens new window)

rocketmq基础
rabbitmq理论到实战

← rabbitmq理论到实战→

最近更新
01
国际象棋
09-15
02
成语
09-15
03
自然拼读
09-15
更多文章>
Theme by Vdoing | Copyright © 2019-2023 zxpnet | 粤ICP备14079330号-1