RocketMQ源码本地搭建调试
Posted JavaEdge.
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMQ源码本地搭建调试相关的知识,希望对你有一定的参考价值。
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,报错如下:
提示在我们配置的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 总结
中间出一些问题,都能根据提示信息在源码中找到相应解决方案,多看源码,是成长捷径。
以上是关于RocketMQ源码本地搭建调试的主要内容,如果未能解决你的问题,请参考以下文章