RocketMQ 源碼本地搭建調試指南

1 GitHub 源碼

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

導入 IDEA，可在命令行執行 mvn compile 一下，保證源碼能夠正確編譯。

2 啓動 Namesrv

到 namesrv 模塊找到 NamesrvStartup，啓動 main 方法，報錯如下：

Please set the ROCKETMQ_HOME variable in your environment to match the location of the RocketMQ installation

需設置環境變量ROCKETMQ_HOME。怎麼設置呢？看看報錯位置：

// org.apache.rocketmq.namesrv.NamesrvStartup#createNamesrvController
if (null == namesrvConfig.getRocketmqHome()) {
    System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
    System.exit(-2);
}

從 namesrvConfig.getRocketmqHome 中獲取的 ROCKETMQ_HOME，進一步發現是 NamesrvConfig 裏面的一個屬性

// org.apache.rocketmq.common.namesrv.NamesrvConfig
// ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir";
// ROCKETMQ_HOME_ENV = "ROCKETMQ_HOME";
private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
public static void main(String[] args) {
    System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, "/Users/apple/doc/IDEAProjects/rocketmq");
    main0(args);
}

再次啓動 Namesrv，報錯如下：

圖片

提示在我們配置的 ROCKETMQ_HOME 目錄下，沒找到 conf\logback_namesrv.xml 配置文件。

在項目中搜索 logback_namesrv.xml 文件，找到在 distribution\conf 目錄下存在很多配置，可直接使用這些配置文件。直接把 distribution\conf 目錄 copy 到配置的 ROCKETMQ_HOME 目錄下即可。 然後在來啓動一下，控制檯打印：

圖片

說明 NameSrv 啓動成功

3 啓動 Broker

找到 BrokerStartup，直接啓動：

Please set the ROCKETMQ_HOME variable in your environment to match the location of the RocketMQ installationDisconnected from the target VM, address: '127.0.0.1:51797', transport: 'socket'

知道咋辦了吧。

broker 需要連接 namrsrv，所以在 BrokerStartup 的 main 方法第一行添加如下兩行代碼：

public static void main(String[] args) {
   System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, "/Users/apple/doc/IDEAProjects/rocketmq");
        System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:9876");
    start(createBrokerController(args));
}

再啓動：

圖片

雖然打印啓動成功，再驗證是否能成功發送和接收消息。

4 啓動 Producer

Producer 的 demo 代碼：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class ProducerTest {
    public static void main(String[] args) throws Exception {
        String namesrvAddr = "127.0.0.1:9876";
        String group = "test_group";
        String topic = "test_hello_rocketmq";
        // 構建Producer實例
        DefaultMQProducer producer = new DefaultMQProducer();
        producer.setNamesrvAddr(namesrvAddr);
        producer.setProducerGroup(group);
        // 啓動producer
        producer.start();
        // 發送消息
        SendResult result = producer.send(new Message(topic, "hello rocketmq".getBytes()));
        System.out.println(result.getSendStatus());
        // 關閉producer
        producer.shutdown();
    }
}

啓動 ProducerTest，控制檯打印SEND_OK，我們的 producer 發送消息也 OK。

5 啓動 Consumer

上代碼：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class ConsumerTest {

    public static void main(String[] args) throws Exception {
        String namesrvAddr = "127.0.0.1:9876";
        String group = "test_consumer_group";
        String topic = "test_hello_rocketmq";
        // 初始化consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumerGroup(group);
        // 訂閱topic
        consumer.subscribe(topic, (String) null);
        // 設置消費的位置，由於producer已經發送了消息，所以我們設置從第一個開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 添加消息監聽器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                msgs.forEach(msg -> {
                    System.out.println(new String(msg.getBody()));
                });
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 啓動consumer
        consumer.start();
        // 由於是異步消費，所以不能立即關閉，防止消息還未消費到
        TimeUnit.SECONDS.sleep(2);
        consumer.shutdown();
    }
}

啓動消費者，能夠成功消費到消息，控制檯打印hello rocketmq。

6 總結

中間出一些問題，都能根據提示信息在源碼中找到相應解決方案，多看源碼，是成長捷徑。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/6Ks3xzXinJRuuNA_84moxA

猜你喜歡