Linux下RocketMQ单机安装

1、优化

主机名、防火墙关闭、tcp优化等

2、jdk安装（1.8）

wget ftp://10.249.100.249/pub/jdk/jdk8/*

3、Maven 3.0.5

打包过程省了，开发给我打包好

4、下载rocketmq

wget https://github.com/apache/rocketmq/archive/rocketmq-all-4.2.0.tar.gz

5、环境变量

vi /etc/profile

export rocketmq=/usr/local/rocketmq-4-3-0

export PATH=$PATH:$rocketmq/bin

source /etc/profile

6、启动mqnameserver

配置内存

/usr/local/rocketmq-4-3-0/bin

runserver.sh

runbroker.sh

测试环境给1g即可

/usr/local/rocketmq-4-3-0/usr/local/rocketmq-4-3-0/bin目录下

cd /var/log/

touch rocketmq.log

nohup sh mqnamesrv > /var/log/rocketmq.log 2>&1 &

nohup: ignoring input

Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release

Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.

The Name Server boot success. serializeType=JSON

7、启动mqbroker

启动mqbroker(/usr/local/rocketmq-4-3-0/bin目录下)

nohup sh mqbroker -n localhost:9876 autoCreateTopicEnable=true > /var/log/rocketmq-broker.log 2>&1 &

注意：localhost可以换成你刚才启动mqnamesrv的ip

[root@ip-10-101-200-47 bin]# jps

25105 BrokerStartup

25045 NamesrvStartup

25158 Jps



8、维护

关闭服务器与启动服务的顺序是相反的

停止broker

sh  mqshutdown broker

停止namerserver

sh  mqshutdown namesrv

[root@ip-10-101-200-47 bin]# sh  mqshutdown broker

The mqbroker(25105) is running...

Send shutdown request to mqbroker(25105) OK

[root@ip-10-101-200-47 bin]# sh  mqshutdown namesrv

The mqnamesrv(25045) is running...

Send shutdown request to mqnamesrv(25045) OK

[2]+  Exit 143                nohup sh mqbroker -n localhost:9876 autoCreateTopicEnable=true > /var/log/rocketmq-broker.log 2>&1

9、图形界面上传启动

nohup java -jar rocketmq-console-ng-1.0.0.jar  > /var/log/rocketmq-console-ng 2>&1 &

http://10.101.200.47:8080/#/

10、监控上传启动

nohup java -jar zipkin.jar  > /var/log/zipkin.log 2>&1 &

http://10.101.200.47:9411/zipkin/



================================

常用命令参考文档https://blog.csdn.net/varyall/article/details/79630037

首先进入 RocketMQ 工程，进入/RocketMQ/bin   在该目录下有个 mqadmin 脚本 .

查看帮助：   在 mqadmin 下可以查看有哪些命令  

查看具体命令的使用

   sh mqadmin   

命令名称 

   sh mqadmin help  

查看 updateTopic 的使用

   sh mqadmin help updateTopic 

关闭nameserver和所有的broker:

   进入到bin下： 

   sh mqshutdown namesrv

   sh mqshutdown broker

   

查看所有消费组group:

   sh mqadmin consumerProgress -n 10.101.200.47:9876

   

查看指定消费组下的所有topic数据堆积情况：

   sh mqadmin consumerProgress -n 10.101.200.47:9876 -g warning-group

    

查看所有topic :

   sh mqadmin topicList -n 10.101.200.47:9876

查看topic信息列表详情统计

   sh mqadmin topicstatus -n 10.101.200.47:9876 -t topicWarning

   

新增topic

   sh mqadmin updateTopic –n 10.101.200.47:9876 –c DefaultCluster –t topicWarning

   

删除topic

   sh mqadmin deleteTopic –n 10.101.200.47:9876 –c DefaultCluster –t topicWarning

 

查询集群消息

   sh mqadmin  clusterList -n 10.101.200.47:9876

====================================

windows

启动

1、进入bin目录：双击启动mqnamesrv.cmd，保持mqnamesrv.exe运行，不要关闭这个终端。

2、另外起一个cmd窗口进入bin目录输入：mqbroker -n 127.0.0.1:9876

linux

安装

解压即可，jdk1.8以上。

启动

1、nohup sh mqnamesrv > /dev/null 2>&1 &

2、nohup sh mqbroker -n 10.10.103.67:9876 > /dev/null 2>&1 &



关闭

1、sh bin/mqshutdown broker

2、sh bin/mqshutdown namesrv

单靠脑袋是记不住的鸭~

文章学习总结于视频：B站学习视频

视频中有一些错误，已经在文章中修改过来了

这篇文章主要学习RocketMQ的下载、安装、启动、关闭这些基础的流程。

文章目录
1. 准备工作
1.1 下载RocketMQ
2. 环境要求
3. 安装步骤
4. 目录介绍
5. 启动RocketMQ
6. 测试RocketMQ
6.1  发送消息
6.2 接收消息
7.  关闭RocketMQ

1. 准备工作
1.1 下载RocketMQ
点我下载，下载的时候下载二进制版本。
2. 环境要求

Linux64位系统
JDK1.8(64位)

3. 安装步骤
本教程以二进制包方式安装

解压安装包
进入安装目录
我喜欢将文件夹名字修改得简单点，并且移动到 /usr/local/ 目录下。

/usr/local/rocketmq

4. 目录介绍

bin：启动脚本，包括shell脚本和CMD脚本
conf：实例配置文件 ，包括broker配置文件、logback配置文件等
lib：依赖jar包，包括Netty、commons-lang、FastJSON等

5. 启动RocketMQ

启动NameServer

# 1.启动NameServer
nohup sh bin/mqnamesrv &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log


启动Broker

# 1.启动Broker
nohup sh bin/mqbroker -n localhost:9876 &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log 



问题描述：
RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小，它们在 bin 目录下：


# 编辑runbroker.sh和runserver.sh修改默认JVM大小
vi runbroker.sh
vi runserver.sh


参考设置：

JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
6. 测试RocketMQ
6.1  发送消息
# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.使用安装包的Demo发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

6.2 接收消息
# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

7.  关闭RocketMQ
# 1.关闭NameServer
sh bin/mqshutdown namesrv
# 2.关闭Broker
sh bin/mqshutdown broker


技 术 无 他， 唯 有 熟 尔。

知 其 然， 也 知 其 所 以 然。

踏 实 一 些， 不 要 着 急， 你 想 要 的 岁 月 都 会 给 你。

目录

Linux下安装RocketMQ

准备工作

安装

第一步、上传文件，解压

第二步、配置环境变量，启动rocketmq

第三步、关闭服务

测试

第一步、打开相应端口

第二步、修改broker.conf配置文件

第三步、编写代码进行测试

Linux下安装RocketMQ

准备工作

RocketMQ下载地址：http://rocketmq.apache.org/release_notes/              选择对应的版本号进行下载，我下载的是4.7.1

带bin的是二进制文件，带source的是源码

中文官方文档：https://github.com/apache/rocketmq/tree/master/docs/cn

官方推荐配置：

64bit OS, Linux/Unix/Mac is recommended;(Windows user see guide below)
	
64bit JDK 1.8+;
	
Maven 3.2.x;
	
Git;
	
4g+ free disk for Broker server
我的linux版本：CentOS 7



 

安装rocketmq前需要先安装jdk1.8，因为rocketmq是Java语言写的

JDK1.8安装教程：https://blog.csdn.net/Me_xuan/article/details/111306026

 

安装

第一步、上传文件，解压

将我们下载的zip的包上传至服务器，新建文件夹，将zip包复制到我们新建的文件夹下面，然后进行解压，我的目录是/app/rocketmq



解压完后重命名，也可以不重命名，因为之前的名字太长，建议还是重命名一下比较好



 

第二步、配置环境变量，启动rocketmq

配置环境变量

vi /etc/profile

# rocketmq

export ROCKETMQ_HOME=/app/rocketmq/rocketmq

export PATH=$PATH:${ROCKETMQ_HOME}/bin

ROCKETMQ_HOME就是我们刚安装的racketmq的地址



配置完环境变量，我们启动一下NameServer

nohup sh mqnamesrv &



如果没有配置环境变量。则得在mq的bin目录下执行 sh mqnamesrv



如果此时启动不起来，看一下rockermq设置的jvm参数，可以把-Xms -Xmx -Xmn 这三个参数根据自己的服务器配置进行相应的调整



启动成功后查看rocketmq的进程



也可以看打印的日志  tail -f ~/logs/rocketmqlogs/namesrv.log





启动borker

先根据自己的服务器配置修改JVM参数

vi runbroker.sh



启动

nohup sh mqbroker -n localhost:9876 &

-n参数后面接要连接到的NameServer的ip和端口 ， ip:port;ip:port 多个用分号隔开。

日志打印





至此，我们的rocketmq服务端就已经启动成功了。

 

第三步、关闭服务

NameServer和broker关闭

 # 关闭NameServer

sh mqshutdown namesrv

 # 关闭broker

sh mqshutdown broker



 

测试

第一步、打开相应端口

开放9876、10911端口

查看防火墙状态：[root@localhost ~]# systemctl status firewalld

查看开放端口：[root@localhost ~]# firewall-cmd --list-all

开放端口：

firewall-cmd --zone=public --add-port=9876/tcp --permanent

firewall-cmd --zone=public --add-port=10911/tcp --permanent

刷新防火墙：firewall-cmd --reload



注意，这里有个坑

之前我天真的以为打开这个端口9876就可以了，结果还是不行，然后得再打开10911这个端口

如果不打开10911这个端口，生产者服务会报超时异常，然后一直打印[NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[] result: true

如：

RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap).

RocketMQLog:WARN Please initialize the logger system properly.

Exception in thread "main" org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout

    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:667)

    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1343)

    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1289)

    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:325)

    at com.example.demo.rocketmq.producer.SyncProducer.main(SyncProducer.java:30)

23:18:22.026 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[] result: true

23:18:52.508 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[] result: true

23:19:22.510 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[] result: true



通过日志可以看到，broker使用了10911这个端口

因为代理服务器(Broker Server)在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备，所以broker的端口也得打开





然后再启动生产者代码，会正常打印相关日志，address中会有我们的两个服务地址，如下

00:01:14.782 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.168.75.100:10911] result: true

00:01:14.785 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.168.75.100:9876] result: true

 

第二步、修改broker.conf配置文件

需要配置broker.conf 这个配置文件，不然的话自己的java项目是无法连接至服务端的

[root@localhost bin]# vi ../conf/broker.conf

增加两个配置项：

# 当前服务器ip:port

namesrvAddr=192.168.75.100:9876

# 当前服务器ip

brokerIP1=192.168.75.100



然后重新启动NameServer和broker

nohup sh mqnamesrv -n 192.168.75.100:9876 &

nohup sh mqbroker -n 192.168.75.100:9876 -c ../conf/broker.conf autoCreateTopicEnable=true &

autoCreateTopicEnable=true 的作用是自动创建topic

 

第三步、编写代码进行测试

代码示例：

官方文档中有简单案例：

https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

我这就是直接拷的官方demo

我这个是springboot的项目

1.引入maven相关依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>

2.编写生产者代码

package com.example.demo.rocketmq.producer;

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.common.RemotingHelper;

/**
 * @ClassName SyncProducer
 * @Author chenxuan
 * @Date 2020/12/15 23:11
 * @Version 1.0
 */
public class SyncProducer {

    public static void main(String[] args) throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("192.168.75.100:9876");
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 10; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
        }
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }
}


3.编写消费者代码

package com.example.demo.rocketmq.consumer;

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.message.MessageExt;

import java.util.List;

/**
 * @ClassName Consumer
 * @Author chenxuan
 * @Date 2020/12/15 23:11
 * @Version 1.0
 */
public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");

        // 设置NameServer的地址
        consumer.setNamesrvAddr("192.168.75.100:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                msgs.stream().forEach(msg ->
                    System.out.println("msg:"+new String(msg.getBody()))
                );
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}


4.启动代码，成功运行

生产者日志：

RocketMQLog:WARN Please initialize the logger system properly.

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E3E80000, offsetMsgId=C0A84B6400002A9F000000000000178E, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=9]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E3FB0001, offsetMsgId=C0A84B6400002A9F0000000000001857, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=2], queueOffset=7]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E3FF0002, offsetMsgId=C0A84B6400002A9F0000000000001920, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=3], queueOffset=6]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4010003, offsetMsgId=C0A84B6400002A9F00000000000019E9, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=8]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4050004, offsetMsgId=C0A84B6400002A9F0000000000001AB2, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=10]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4090005, offsetMsgId=C0A84B6400002A9F0000000000001B7B, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=2], queueOffset=8]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E40C0006, offsetMsgId=C0A84B6400002A9F0000000000001C44, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=3], queueOffset=7]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4100007, offsetMsgId=C0A84B6400002A9F0000000000001D0D, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=9]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4120008, offsetMsgId=C0A84B6400002A9F0000000000001DD6, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=11]

SendResult [sendStatus=SEND_OK, msgId=C0A8016931CC00B4AAC25266E4140009, offsetMsgId=C0A84B6400002A9F0000000000001E9F, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=2], queueOffset=9]

00:01:14.782 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.168.75.100:10911] result: true

00:01:14.785 [NettyClientSelector_1] INFO RocketmqRemoting - closeChannel: close the connection to remote address[192.168.75.100:9876] result: true

Process finished with exit code 0

消费者日志（只打印了消息内容，不然日志太多）：

Consumer Started.

msg:Hello RocketMQ 0

msg:Hello RocketMQ 1

msg:Hello RocketMQ 2

msg:Hello RocketMQ 3

msg:Hello RocketMQ 4

msg:Hello RocketMQ 5

msg:Hello RocketMQ 6

msg:Hello RocketMQ 7

msg:Hello RocketMQ 8

msg:Hello RocketMQ 9

至此我们的rocketmq安装成功，第一个rocketmq的demo也成功运行，后面就可以开始rocketmq之旅了。