RocketMQ 集羣監控以及Hello World

RocketMQ 目前有兩個版本  alibaba版本和apache版本

1、alibaba版本

tomcat部署：

　　apache-tomcat-7.0.90.tar.gz

　　jdk7

　　虛擬機redhat6.5：192.168.1.201

一、上傳rocketmq-console.war包到tomcat下webapps

#新建目錄
mkdir rocketmq-console
#解壓war文件
unzip rocketmq-console.war -d rocketmq-console
#刪除war
rm -f rocketmq-console.war

二、修改配置  指定nameserveraddr

cd /usr/local/tomcat7/webapps/rocketmq-console/WEB-INF/classes

vim config.properties

rocketmq.namesrv.addr=192.168.1.201:9876;192.168.1.202:9876
throwDone=true

 

三、啓動tomcat

[root@201 bin]# /usr/local/tomcat7/bin/startup.sh
Using CATALINA_BASE:   /usr/local/tomcat7
Using CATALINA_HOME:   /usr/local/tomcat7
Using CATALINA_TMPDIR: /usr/local/tomcat7/temp
Using JRE_HOME:        /usr/local/java/jdk1.7.0_80
Using CLASSPATH:       /usr/local/tomcat7/bin/bootstrap.jar:/usr/local/tomcat7/bin/tomcat-juli.jar
Tomcat started.
[root@201 bin]# jps
2133 Jps
1483 NamesrvStartup
2094 Bootstrap
1529 BrokerStartup

 

四、瀏覽器訪問

http://192.168.1.201:8080/rocketmq-console

 

 

/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rocketmq.quickstart;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

/**
 * Producer，發送消息
 * 
 */
public class Producer 
{
    public static void main(String[] args) throws MQClientException,
            InterruptedException 
    {
        DefaultMQProducer producer = new DefaultMQProducer("QuickStart_Producer");
        producer.setNamesrvAddr("192.168.1.201:9876;192.168.1.202:9876");
        producer.start();

        for (int i = 0; i < 1000; i++) 
        {
            try 
            {
                Message msg = new Message("TopicQuickStart",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes()// body
                );
                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        producer.shutdown();
    }
}

/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rocketmq.quickstart;

import java.io.UnsupportedEncodingException;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * Consumer，訂閱消息
 */
public class Consumer {

    public static void main(String[] args) throws InterruptedException,
            MQClientException 
    {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("QuickStart_Consumer");
        consumer.setNamesrvAddr("192.168.1.201:9876;192.168.1.202:9876");
        /**
         * 設置Consumer第一次啓動是從隊列頭部開始消費仍是隊列尾部開始消費<br>
         * 若是非第一次啓動，那麼按照上次消費的位置繼續消費
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicQuickStart", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    for (int i = 0; i < msgs.size(); i++) {
                        MessageExt msg = msgs.get(i);
                        String topic = msg.getTopic();
                        String tag = msg.getTags();
                        String body = new String(msg.getBody(), "UTF-8");
                        System.out.println(Thread.currentThread().getName()
                                + " Receive New Messages: topic=" 
                                + topic+",tag="
                                + tag +",body="
                                + body);

                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        System.out.println("Consumer Started.");
    }
}

生產消息：

消費消息：

監控消息：

 

2、apache版本

貢獻給apache以後，上面的那個工具就沒法使用了，不過今天從網上找到了個新的管理界面。

github地址：https://github.com/apache/incubator-rocketmq-externals

2018年1月23日，更新最新github地址爲：https://github.com/apache/rocketmq-externals

 

 

參照幫助文件使用便可：

幫助文檔路徑：https://github.com/apache/incubator-rocketmq-externals/blob/master/rocketmq-console/README.md

 

具體以下：

一、修改配置文件，使管理界面與rocketmq集羣產生關聯。

incubator-rocketmq-externals-master/rocketmq-console/src/main/resources/application.properties

修改內容及修改結果以下圖所示：

 

二、編譯rocketmq-console

編譯命令：mvn clean package -Dmaven.test.skip=true（注意：不要直接使用mvn package，會提示不少錯誤）

 

三、將編譯好的jar包上傳到linux服務器

（若是直接在Linux環境上編譯，能夠省略這步）

 

我這裏上傳到了本地虛擬機192.168.6.5上。路徑爲：/home/hadmin/jar

 

四、運行jar包

命令：java -jar target/rocketmq-console-ng-1.0.0.jar

 

五、使用瀏覽器訪問管理界面

方位地址：http://192.168.6.5:8080/

 

六、可能遇到的問題

畫面能夠正常啓動，不過從控制檯的監控日誌上看，存在以下的錯誤日誌。

org.apache.rocketmq.remoting.exception.RemotingTimeoutException: wait response on the channel <192.168.1.80:10918> timeout, 5000(ms)

緣由是isVIPChannel默認爲true，會監控rocketmq的vip通道，將該屬性設置爲false便可。

設置後的配置文件以下所示：

server.contextPath=
server.port=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.1.80:9876;192.168.1.81:9876
#if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
rocketmq.config.isVIPChannel=false
#rocketmq-console's data path:dashboard/monitor
rocketmq.config.dataPath=/home/hadmin/data/rocketmq
#set it false if you don't want use dashboard.default true
rocketmq.config.enableDashBoardCollect=true

 

參考：https://www.cnblogs.com/quchunhui/p/7284752.html