Linux 从零搭建RocketMQ及进行可视化的监控和管理

系统管理员 2022-12-04 04:28 317阅读 0赞

### 文章目录 ###

*   *  说明
     *  搭建
     *  测试 RocketMQ
     *  构建RocketMQ监控

## 说明 ##

*  由于我的服务器内存很小，所以搭建暂时不使用 dledger构建集群

## 搭建 ##

本次搭建仅搭建一个NameServer和一个Broker  
[官方搭建文档][Link 1]

我这里按我的步骤来及一些坑分享给搭建,这里是以源码方式构建，也可以直接下载二进制包进行构建

1.  配置Java 环境变量
2.  构建 DLedger

git clone https://github.com/openmessaging/openmessaging-storage-dledger.git

cd openmessaging-storage-dledger

mvn clean install -DskipTests

1.  构建 RocketMQ

git clone https://github.com/apache/rocketmq.git

cd rocketmq

git checkout -b store_with_dledger origin/store_with_dledger

mvn -Prelease-all -DskipTests clean install -U

1.  快速部署

这里坑就开始来了

首先是启动NameServer

进入到 安装的 distribution目录  
`/app/rocketmq/distribution`

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

发现报错，启动脚本设置JVM内存过大，2g还是3g忘了，修改 runserver.sh 启动脚本JVM参数

vi runserver.sh

这里我设置的堆内存暂时设置为256g  
![在这里插入图片描述][202009051826572.png]  
参考设置：  
`JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"`

同理设置 runbroker.sh启动参数也是如此

然后启动 NameServer 和 Broker

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

使用jps 可以看到 NameServer 和 Broker都启动了  
![在这里插入图片描述][20200905183113395.png]

## 测试 RocketMQ ##

*  发送消息

# 1.设置环境变量
    export NAMESRV_ADDR=localhost:9876
    # 2.使用安装包的Demo发送消息
    sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

*  发送消息

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

发现是可以正常发送消息的

然后编写简单的生产者，消费者代码测试

*  Consumer

public static void main(String[] args) throws Exception{ 
            final DefaultMQProducer producer = new DefaultMQProducer("test_product");
            // 配置NameServer 地址
            producer.setNamesrvAddr("49.243.149.105:9876");
            producer.start();
            //死循环发送消息
            for (int i = 0; i < 10; i++) { 
                Thread thread = new Thread(() -> { 
                    while (true) { 
                        try { 
                            Message message = new Message("TopicTest", "Taga",
                                    "Test".getBytes(RemotingHelper.DEFAULT_CHARSET));
                            SendResult sendResult = producer.send(message);
                            // 通过sendResult返回消息是否成功送达
                            System.out.printf("%s%n", sendResult);
                            Thread.sleep(1000);
                        } catch (Exception e) { 
                            e.printStackTrace();
                        }
                    }
    
                });
                thread.start();
                // 不让线程结束，一直发消息
                thread.join();
    
            }
    
        }

*  Producer

public static void main(String[] args) throws Exception{ 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer");
            // 设置NameServer地址
            consumer.setNamesrvAddr("49.243.149.105:9876");
            // 订阅Topic
            consumer.subscribe("TopicTest", "*");
            // 注册回调接口，获取消息
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext consumeConcurrentlyContext) { 
                    System.out.println("接受到的消息为：" + msgs);
                    msgs.forEach(s -> System.out.println(new String(s.getBody())));
    
    
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            consumer.start();
    
        }

> 启动生产者报错：org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout

发现好像是连接不上服务器的Mq，google发现启动脚本需要改为服务器公网IP

然后需要关闭 RocketMQ

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

带上IP 的启动脚本

## 启动NameServer
    nohup ./bin/mqnamesrv -n 49.243.149.105:9876 &
    ## 启动broker
    ## 在conf/broker.conf 中 加入 brokerIP1=你的公网IP
    nohup sh bin/mqbroker -n 49.243.149.105:9876 -c conf/broker.conf autoCreateTopicEnable=true &

然后重新启动测试程序  
![在这里插入图片描述][20200905184500594.png]  
![在这里插入图片描述][20200905184518788.png]

自此简单的RocketMQ集群搭建完成，然后我们来搭建 RocketMQ监控，这个就很容易了

## 构建RocketMQ监控 ##

下载并编译打包

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

启动RocketMQ监控

java -jar rocketmq-console-ng-2.0.0.jar --server.port=8080 --rocketmq.config.namesrvAddr=49.243.149.105:9876

> 注意这里需要制定 NameServer地址，多个NameServer以逗号隔开，我们也可以直接在rocketmq-console的配置文件里面修改中NameServer地址，然后打包编译

![在这里插入图片描述][20200905185030664.png_pic_center]  
![在这里插入图片描述][20200905185124428.png]  
启动完成后我们访问 http://49.243.149.105:8080/  
![在这里插入图片描述][20200905185311860.png]

这里可以切换为我们熟悉的中文  
![在这里插入图片描述][20200905185504168.png]

至此 Mq的集群及监控就完成了!

[Link 1]: https://github.com/apache/rocketmq/blob/master/docs/cn/dledger/quick_start.md
[202009051826572.png]: /images/20221123/96d048d50d6d4aefbaed1b66aaa14059.png
[20200905183113395.png]: /images/20221123/ca35a6d486fc4436a89f9eb56530ce6e.png
[20200905184500594.png]: /images/20221123/47f11d50a41d48deb5c03d58cb7fac07.png
[20200905184518788.png]: /images/20221123/dd8cc0bd6cd14a148a1c169eb3b9289b.png
[20200905185030664.png_pic_center]: /images/20221123/bb85e4b62331426884dcea3882c7ada8.png
[20200905185124428.png]: /images/20221123/b10729290dc549568cb020e897baf9ae.png
[20200905185311860.png]: /images/20221123/77843ecfb3ff496690776e8d53f00d44.png
[20200905185504168.png]: /images/20221123/fa35d3ee23f94c04976497746d441e69.png