RocketMQ源码本地搭建调试

以你之姓@ 2024-03-27 09:24 127阅读 0赞

### 1 GitHub源码 ###

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

导入IDEA，可在命令行执行*mvn compile*一下，保证源码能够正确编译。本次我使用的master分支的版本-4.8.0。下面我们开始准备启动Namesrv。

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

*  为方便，可直接在源码NamesrvStartup的main方法第一行
 *  也可设置环境变量ROCKETMQ\_HOME
 *  或启动时加入命令行参数 -Drocketmq.home.dir=/path

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

再次启动Namesrv，报错如下：

![a757d9e0dd6c446e8f0807790c9bc3ae.png][]

提示在我们配置的*ROCKETMQ\_HOME* 目录下，没找到*conf\\logback\_namesrv.xml* 配置文件。

在项目中搜索*logback\_namesrv.xml* 文件，找到在*distribution\\conf* 目录下存在很多配置，可直接使用这些配置文件。直接把*distribution\\conf* 目录copy到配置的*ROCKETMQ\_HOME* 目录下即可。  
然后在来启动一下，控制台打印：

![020aeb9bf4064a9a83ee1df980153c65.png][]

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

再启动：  
![6e47e5d6c23a4297aa948aa1549f51eb.png][]

虽然打印启动成功，再验证是否能成功发送和接收消息。

### 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 总结 ###

中间出一些问题，都能根据提示信息在源码中找到相应解决方案，多看源码，是成长捷径。

[a757d9e0dd6c446e8f0807790c9bc3ae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/2bd577129fba47418783c2683e4c9d89.png
[020aeb9bf4064a9a83ee1df980153c65.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/a538faf9872a4332b2abfe74a4c0f246.png
[6e47e5d6c23a4297aa948aa1549f51eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/8f1e69afa3194b79a7103250e0765f05.png