快捷導(dǎo)航

RocketMQ源碼本地搭建調(diào)試方法

更新時(shí)間：2023年01月28日 09:49:32 作者：JavaEdge.

這篇文章主要介紹了RocketMQ源碼本地搭建調(diào)試,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

1 GitHub源碼

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

導(dǎo)入IDEA，可在命令行執(zhí)行mvn compile一下，保證源碼能夠正確編譯。本次我使用的master分支的版本-4.8.0。下面我們開(kāi)始準(zhǔn)備啟動(dòng)Namesrv。

2 啟動(dòng)Namesrv

到namesrv模塊找到NamesrvStartup，啟動(dòng)main方法，報(bào)錯(cuò)如下：

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

需設(shè)置環(huán)境變量ROCKETMQ_HOME。怎么設(shè)置呢？看看報(bào)錯(cuò)位置：

// 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，進(jìn)一步發(fā)現(xiàn)是NamesrvConfig里面的一個(gè)屬性

// 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));

為方便，可直接在源碼NamesrvStartup的main方法第一行
也可設(shè)置環(huán)境變量ROCKETMQ_HOME
或啟動(dòng)時(shí)加入命令行參數(shù) -Drocketmq.home.dir=/path

public static void main(String[] args) {
    System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, "/Users/apple/doc/IDEAProjects/rocketmq");
    main0(args);
}

再次啟動(dòng)Namesrv，報(bào)錯(cuò)如下：

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

在項(xiàng)目中搜索logback_namesrv.xml 文件，找到在distribution\conf 目錄下存在很多配置，可直接使用這些配置文件。直接把distribution\conf 目錄copy到配置的ROCKETMQ_HOME 目錄下即可。
然后在來(lái)啟動(dòng)一下，控制臺(tái)打?。?/p>

說(shuō)明NameSrv啟動(dòng)成功

3 啟動(dòng)Broker

找到BrokerStartup，直接啟動(dòng)：

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));
}

再啟動(dòng)：

雖然打印啟動(dòng)成功，再驗(yàn)證是否能成功發(fā)送和接收消息。

4 啟動(dòng)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";
        // 構(gòu)建Producer實(shí)例
        DefaultMQProducer producer = new DefaultMQProducer();
        producer.setNamesrvAddr(namesrvAddr);
        producer.setProducerGroup(group);
        // 啟動(dòng)producer
        producer.start();
        // 發(fā)送消息
        SendResult result = producer.send(new Message(topic, "hello rocketmq".getBytes()));
        System.out.println(result.getSendStatus());
        // 關(guān)閉producer
        producer.shutdown();
    }
}

啟動(dòng)ProducerTest，控制臺(tái)打印SEND_OK，我們的producer發(fā)送消息也OK。

5 啟動(dòng)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);
        // 設(shè)置消費(fèi)的位置，由于producer已經(jīng)發(fā)送了消息，所以我們?cè)O(shè)置從第一個(gè)開(kāi)始消費(fèi)
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 添加消息監(jiān)聽(tīng)器
        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;
            }
        });
        // 啟動(dòng)consumer
        consumer.start();
        // 由于是異步消費(fèi)，所以不能立即關(guān)閉，防止消息還未消費(fèi)到
        TimeUnit.SECONDS.sleep(2);
        consumer.shutdown();
    }
}

啟動(dòng)消費(fèi)者，能夠成功消費(fèi)到消息，控制臺(tái)打印hello rocketmq。