RocketMQ源码本地搭建调试

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

1 GitHub源码

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

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

2 启动Namesrv

到namesrv模块找到NamesrvStartup,启动main方法,报错如下:

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

需设置环境变量ROCKETMQ_HOME。怎么设置呢?看看报错位置:

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

从namesrvConfig.getRocketmqHome中获取的ROCKETMQ_HOME,进一步发现是NamesrvConfig里面的一个属性

  1. // org.apache.rocketmq.common.namesrv.NamesrvConfig
  2. // ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir";
  3. // ROCKETMQ_HOME_ENV = "ROCKETMQ_HOME";
  4. 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) {

    1. System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, "/Users/apple/doc/IDEAProjects/rocketmq");
    2. 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,直接启动:

  1. 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方法第一行添加如下两行代码:

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

再启动:
6e47e5d6c23a4297aa948aa1549f51eb.png

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

4 启动Producer

Producer的demo代码:

  1. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  2. import org.apache.rocketmq.client.producer.SendResult;
  3. import org.apache.rocketmq.common.message.Message;
  4. public class ProducerTest {
  5. public static void main(String[] args) throws Exception {
  6. String namesrvAddr = "127.0.0.1:9876";
  7. String group = "test_group";
  8. String topic = "test_hello_rocketmq";
  9. // 构建Producer实例
  10. DefaultMQProducer producer = new DefaultMQProducer();
  11. producer.setNamesrvAddr(namesrvAddr);
  12. producer.setProducerGroup(group);
  13. // 启动producer
  14. producer.start();
  15. // 发送消息
  16. SendResult result = producer.send(new Message(topic, "hello rocketmq".getBytes()));
  17. System.out.println(result.getSendStatus());
  18. // 关闭producer
  19. producer.shutdown();
  20. }
  21. }

启动ProducerTest,控制台打印SEND_OK,我们的producer发送消息也OK。

5 启动Consumer

上代码:

  1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  2. import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
  3. import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
  4. import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
  5. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  6. import org.apache.rocketmq.common.message.MessageExt;
  7. import java.util.List;
  8. import java.util.concurrent.TimeUnit;
  9. public class ConsumerTest {
  10. public static void main(String[] args) throws Exception {
  11. String namesrvAddr = "127.0.0.1:9876";
  12. String group = "test_consumer_group";
  13. String topic = "test_hello_rocketmq";
  14. // 初始化consumer
  15. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
  16. consumer.setNamesrvAddr(namesrvAddr);
  17. consumer.setConsumerGroup(group);
  18. // 订阅topic
  19. consumer.subscribe(topic, (String) null);
  20. // 设置消费的位置,由于producer已经发送了消息,所以我们设置从第一个开始消费
  21. consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  22. // 添加消息监听器
  23. consumer.registerMessageListener(new MessageListenerOrderly() {
  24. @Override
  25. public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
  26. msgs.forEach(msg -> {
  27. System.out.println(new String(msg.getBody()));
  28. });
  29. return ConsumeOrderlyStatus.SUCCESS;
  30. }
  31. });
  32. // 启动consumer
  33. consumer.start();
  34. // 由于是异步消费,所以不能立即关闭,防止消息还未消费到
  35. TimeUnit.SECONDS.sleep(2);
  36. consumer.shutdown();
  37. }
  38. }

启动消费者,能够成功消费到消息,控制台打印hello rocketmq

6 总结

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

发表评论

表情:
评论列表 (有 0 条评论,227人围观)

还没有评论,来说两句吧...

相关阅读

    相关 调试RocketMQ

    1. 拷贝`namesrv`、`broker`的配置文件到指定目录,为了避免直接修改代码中的配置文件。 1.1 在F盘创建`rocketmq`文件夹,创建三个子文