一、Linux部署
准备一台Linux机器,部署单master rocketmq节点
系统 | ip | 角色 | 模式 |
---|---|---|---|
CENTOS | 10.4.7.126 | Nameserver,broker | Master |
1. 配置JDK
rocketmq运行需要依赖jdk,安装步骤略。
2. 下载和配置
- 从官网下载安装包 https://rocketmq.apache.org/zh/download/ ,并上传到CENTOS系统中,并进行解压
cd /data
# 解压
unzip rocketmq-all-5.1.3-bin-release.zip
# 创建目录软链接
ln -s rocketmq-all-5.1.3-bin-release rocketmq
# 创建目录
# 消息存储及位点目录
mkdir -p /data/rocketmq/store
mkdir -p /data/rocketmq/store/commitlog
mkdir -p /data/rocketmq/store/consumequeue
mkdir -p /data/rocketmq/store/index
# 创建日志目录
mkdir -p /data/rocketmq/logs
# 进入配置文件目录并替换日志配置文件中的日志目录的位置
cd /data/rocketmq/conf && sed -i 's#${user.home}#/data/rocketmq#g' *.xml
修改rocketmq启动所需配置文件
# 进入rocketmq自带的配置文件目录
[root@mq-126 conf]# cd /data/rocketmq/conf/
[root@mq-126 conf]# ls
# 2master 2slave 异步消息同步配置目录
2m-2s-async
# 2master 2slave 同步消息同步配置目录
2m-2s-sync
# 2master no slave 配置目录
2m-noslave
# 单点配置目录
broker.conf
container
controller
dledger
plain_acl.yml
# 日志配置文件
rmq.broker.logback.xml
rmq.client.logback.xml
rmq.controller.logback.xml
rmq.namesrv.logback.xml
rmq-proxy.json
rmq.proxy.logback.xml
rmq.tools.logback.xml
tools.yml
[root@mq-126 conf]# vi broker.conf
将下面的配置赋值到broker.conf中
#所属集群名字
brokerClusterName=rocketmq-cluster
#broker 名字,注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master,>0 表示 Slave
brokerId=0
#nameServer 地址,分号分割
namesrvAddr=127.0.0.1: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=/data/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/data/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/data/rocketmq/store/consumequeue #消息索引存储路径
storePathIndex=/data/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/data/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/data/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
修改启动脚本的jvm参数,找到对应行进行修改,原配置设置的内存太大,我的虚拟机是1g内存的,会无法正常启动,需要改小。
-
vim /data/rocketmq/bin/runbroker.sh
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m"
-
vim /data/rocketmq/bin/runserver.sh
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m"
3. 启动
-
启动nameserver
[root@mq-126 rocketmq]# nohup sh /data/rocketmq/bin/mqnamesrv & [root@mq-126 rocketmq]# jps 2617 NamesrvStartup 2654 Jps [root@mq-126 rocketmq]# netstat -ano | grep 9876 tcp6 0 0 :::9876 :::* LISTEN off (0.00/0/0) tcp6 0 0 10.4.7.126:9876 10.4.7.126:53946 ESTABLISHED off (0.00/0/0) tcp6 0 0 10.4.7.126:9876 10.4.7.125:42368 ESTABLISHED off (0.00/0/0) tcp6 0 0 10.4.7.126:53946 10.4.7.126:9876 ESTABLISHED off (0.00/0/0)
-
启动broker
[root@mq-126 bin]# nohup sh /data/rocketmq/bin/mqbroker -c /data/rocketmq/conf/broker.conf & [2] 2829 [root@mq-126 bin]# jps 2796 NamesrvStartup 2861 BrokerStartup 2893 Jps
-
启动dashboard
启动dashboard最快的方式是通过docker进行启动。
[root@mq-126 bin]# systemctl start docker [root@mq-126 bin]# docker run -d --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.namesrv.addr=10.4.7.126:9876" -p 8080:8080 -t apacherocketmq/rocketmq-dashboard:latest d55cb8b7ff0e9d19964c2019eaf4bba8301cfc2cf1b8a3eb01bff09df134f90e
然后就可以访问dashboard了
如果没有docker环境 在官网下下载源码包,使用maven编译后进行启动
https://rocketmq.apache.org/zh/download/
4. 发送和消费测试代码
Producer
package cn.shutdown.mq;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
/**
* 消息生产者
* @author Domino.J
* @date 2023/8/11
*/
public class Producer {
public static final String NAMESRV_ADDR = "10.4.7.126:9876";
public static void main(String[] args)
throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("test_quick_producer_name");
producer.setNamesrvAddr(NAMESRV_ADDR);
producer.start();
for (int i = 0; i < 1000; i++) {
Message message = new Message("test_quick_topic",//主题
"TagA", //标签
"KeyA" + i, //用户自定义的key,唯一的标识
("Hello RocketMQ" + i).getBytes());//消息内容
//发送消息
SendResult sr = producer.send(message);
System.out.println("消息发送结果:" + sr);
Thread.sleep(1000);
}
producer.shutdown();
}
}
Consumer
package cn.shutdown.mq;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.util.List;
/**
* 消息消费者
* @author Domino.J
* @date 2023/8/11
*/
public class Consumer {
public static final String NAMESRV_ADDR = "10.4.7.126:9876";
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
consumer.setNamesrvAddr(NAMESRV_ADDR);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.subscribe("test_quick_topic", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
MessageExt me = msgs.get(0);
try {
String topic = me.getTopic();
String tags = me.getTags();
String keys = me.getKeys();
String msgBody = new String(me.getBody(), RemotingHelper.DEFAULT_CHARSET);
System.err.println("topic: " + topic + ",tags: " + tags + ", keys: " + keys + ",body: " + msgBody);
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.err.println("consumer start...");
}
}
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.shutdown</groupId>
<artifactId>rocketmq-demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>rocketmq-demo</name>
<description>rocketmq-demo</description>
<properties>
<java.version>1.8</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spring-boot.version>2.6.13</spring-boot.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client-java</artifactId>
<version>5.0.3</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>5.1.3</version>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring-boot.version}</version>
<configuration>
<mainClass>cn.shutdown.mq.RocketmqDemoApplication</mainClass>
<skip>true</skip>
</configuration>
<executions>
<execution>
<id>repackage</id>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
运行结果
二、Docker Compose部署
1. 拉取镜像
拉取 rocketmq镜像
docker pull apache/rocketmq:5.1.3
拉取 rocketmq dashboard镜像
docker pull apacherocketmq/rocketmq-dashboard:latest
2. 创建挂载目录,拷贝挂载文件
-
创建Nameserver需要的挂载目录并拷贝挂载文件
# 日志目录 mkdir /data/opt/rocketmq/nameserver/logs -p # 脚本目录 mkdir /data/opt/rocketmq/nameserver/bin -p
设置权限:如果不设置会导致NameServer容器内部无法写日志文件
# 777 文件所属者、文件所属组和其他人有读取 & 写入 & 执行全部权限。rwxrwxrwx chmod 777 -R /data/opt/rocketmq/nameserver/*
正如使用Linux部署一样,需要修改JAVA_OPT配置,这里需要从rocketmq容器中复制出一份运行文件,修改后再挂载到新启动的容器上。另外,NameServer启动脚本中有一个自动计算最大堆内存和新生代内存的函数会导致在不同硬件环境下设置最大堆内存和新生代内存环境变量不被应用,,这里先提前copy一份容器内部启动脚本做挂载,如果想自定义内存可以自己调整。
启动临时rocketmq容器
docker run -d \ --privileged=true \ --name rmqnamesrv \ apache/rocketmq:5.1.3 sh mqnamesrv
复制容器内的文件到挂载目录中
[root@mq-126 bin]# docker cp rmqnamesrv:/home/rocketmq/rocketmq-5.1.3/bin/ /data/opt/rocketmq/nameserver/ [root@mq-126 bin]# ls cachedog.sh mqbroker.numanode0 os.sh cleancache.sh mqbroker.numanode1 play.cmd cleancache.v1.sh mqbroker.numanode2 play.sh controller mqbroker.numanode3 README.md dledger mqcontroller runbroker.cmd docker-entrypoint.sh mqcontroller.cmd runbroker.sh export.sh mqnamesrv runserver.cmd mqadmin mqnamesrv.cmd runserver.sh mqadmin.cmd mqproxy setcache.sh mqbroker mqproxy.cmd startfsrv.sh mqbroker.cmd mqshutdown tools.cmd mqbrokercontainer mqshutdown.cmd tools.sh
修改 runserver.sh,把 calculate_heap_sizes 这句注释掉,
# 注释掉以下这句 # calculate_heap_sizes # Dynamically calculate parameters, for reference.
然后停止并删除容器
docker stop rmqnamesrv && docker rm rmqnamesrv
-
创建Broker需要挂载的文件夹和文件
# 创建需要的挂载目录 mkdir /data/opt/rocketmq/broker/logs -p mkdir /data/opt/rocketmq/broker/data -p mkdir /data/opt/rocketmq/broker/conf -p mkdir /data/opt/rocketmq/broker/bin -p mkdir /data/opt/rocketmq/broker/store -p
给目录赋权限
# 777 文件所属者、文件所属组和其他人有读取 & 写入 & 执行全部权限。rwxrwxrwx chmod 777 -R /data/opt/rocketmq/broker/*
启动临时容器
docker run -d \ --name rmqbroker \ --privileged=true \ apache/rocketmq:5.1.3 \ sh mqbroker
拷贝运行文件
[root@mq-126 bin]# docker cp rmqbroker:/home/rocketmq/rocketmq-5.1.3/bin/ /data/opt/rocketmq/broker/ Successfully copied 110kB to /data/opt/rocketmq/broker/ [root@mq-126 bin]# ls cachedog.sh mqbroker.numanode0 os.sh cleancache.sh mqbroker.numanode1 play.cmd cleancache.v1.sh mqbroker.numanode2 play.sh controller mqbroker.numanode3 README.md dledger mqcontroller runbroker.cmd docker-entrypoint.sh mqcontroller.cmd runbroker.sh export.sh mqnamesrv runserver.cmd mqadmin mqnamesrv.cmd runserver.sh mqadmin.cmd mqproxy setcache.sh mqbroker mqproxy.cmd startfsrv.sh mqbroker.cmd mqshutdown tools.cmd mqbrokercontainer mqshutdown.cmd tools.sh
修改 runbroker.sh,把 calculate_heap_sizes 这句注释掉,
# 注释掉以下这句 # calculate_heap_sizes # Dynamically calculate parameters, for reference.
拷贝配置文件
[root@mq-126 conf]# docker cp rmqbroker:/home/rocketmq/rocketmq-5.1.3/conf/ /data/opt/rocketmq/broker/ Successfully copied 145kB to /data/opt/rocketmq/broker/ [root@mq-126 conf]# ls 2m-2s-async dledger rmq-proxy.json 2m-2s-sync plain_acl.yml rmq.proxy.logback.xml 2m-noslave rmq.broker.logback.xml rmq.tools.logback.xml broker.conf rmq.client.logback.xml tools.yml container rmq.controller.logback.xml controller rmq.namesrv.logback.xml
修改 broker.conf
# 集群名称 brokerClusterName = DefaultCluster # 节点名称 brokerName = broker-a # broker id节点ID, 0 表示 master, 其他的正整数表示 slave,不能小于0 brokerId = 0 # Broker服务地址 String 内部使用填内网ip,如果是需要给外部使用填公网ip brokerIP1 = 10.4.7.125 # Broker角色 brokerRole = ASYNC_MASTER # 刷盘方式 flushDiskType = ASYNC_FLUSH # 在每天的什么时间删除已经超过文件保留时间的 commit log,默认值04 deleteWhen = 04 # 以小时计算的文件保留时间 默认值72小时 fileReservedTime = 72 # 是否允许Broker 自动创建Topic,建议线下开启,线上关闭 autoCreateTopicEnable=true # 是否允许Broker自动创建订阅组,建议线下开启,线上关闭 autoCreateSubscriptionGroup=true
停止并删除临时的broker容器
docker stop rmqbroker && docker rm rmqbroker
3. 准备Docker Compose 文件
在/data/opt/rocketmq
目录创建docker-compose.yml文件
version: '3.8'
services:
rmqnamesrv:
image: apache/rocketmq:5.1.3
container_name: rmqnamesrv
ports:
- 9876:9876
restart: always
privileged: true
volumes:
- /data/opt/rocketmq/nameserver/logs:/home/rocketmq/logs
- /data/opt/rocketmq/nameserver/bin/runserver.sh:/home/rocketmq/rocketmq-5.1.3/bin/runserver.sh
environment:
- MAX_HEAP_SIZE=256M
- HEAP_NEWSIZE=128M
command: ["sh","mqnamesrv"]
broker:
image: apache/rocketmq:5.1.3
container_name: rmqbroker
ports:
- 10909:10909
- 10911:10911
- 8081:8081
restart: always
privileged: true
volumes:
- /data/opt/rocketmq/broker/logs:/home/rocketmq/logs
- /data/opt/rocketmq/broker/store:/home/rocketmq/store
- /data/opt/rocketmq/broker/conf/broker.conf:/home/rocketmq/broker.conf
- /data/opt/rocketmq/broker/bin/runbroker.sh:/home/rocketmq/rocketmq-5.1.3/bin/runbroker.sh
depends_on:
- 'rmqnamesrv'
environment:
- NAMESRV_ADDR=rmqnamesrv:9876
- MAX_HEAP_SIZE=512M
- HEAP_NEWSIZE=256M
command: ["sh","mqbroker","-c","/home/rocketmq/broker.conf","--enable-proxy"]
rmqdashboard:
image: apacherocketmq/rocketmq-dashboard:latest
container_name: rocketmq-dashboard-latest
ports:
- 8080:8080
restart: always
privileged: true
depends_on:
- 'rmqnamesrv'
environment:
- JAVA_OPTS= -Xmx256M -Xms256M -Xmn128M -Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false
使用docker compose 命令启动
[root@mq_125 rocketmq]# vi docker-compose.yml
[root@mq_125 rocketmq]# docker compose up -d
[+] Building 0.0s (0/0)
[+] Running 3/3
✔ Container rmqnamesrv Started 0.7s
✔ Container rocketmq-dashboard-latest Started 2.6s
✔ Container rmqbroker Started
4. 发送和消费消息测试
测试代码与linux部署相同