1.环境准备
1.1.依赖工具
- JDK :1.8+
- Maven
- IntelliJ IDEA
1.2.源码拉取
从官方仓库 https://github.com/apache/rocketmq Fork 出属于自己的仓库。为什么要 Fork ?既然开始阅读、调试源码,我们可能会写一些注释,有了自己的仓库,可以进行自由的提交。
使用 IntelliJ IDEA 从 Fork 出来的仓库拉取代码。拉取完成后,Maven 会下载依赖包,可能会花费一些时间,耐心等待下。
在等待的过程中,我来简单说下,搭建调试环境的过程:
- 启动 RocketMQ Namesrv
- 启动 RocketMQ Broker
- 启动 RocketMQ Producer
- 启动 RocketMQ Consumer
最小化的 RocketMQ 的环境,暂时不考虑 Namesrv 集群、Broker 集群、Consumer 集群。
另外,本文使用的 RocketMQ 版本是 4.4.0-SNAPSHOT 。
1.3.启动 RocketMQ Namesrv
打开 org.apache.rocketmq.namesrv.NameServerInstanceTest 单元测试类,参考 #startup() 方法,我们编写 #main(String[] args) 静态方法,代码如下:
public static void main(String[] args) throws Exception {
// NamesrvConfig 配置
final NamesrvConfig namesrvConfig = new NamesrvConfig();
// NettyServerConfig 配置
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(9876); // 设置端口
// 创建 NamesrvController 对象,并启动
NamesrvController namesrvController = new NamesrvController(namesrvConfig, nettyServerConfig);
namesrvController.initialize();
namesrvController.start();
// 睡觉,就不起来
Thread.sleep(DateUtils.MILLIS_PER_DAY);
}
然后,右键运行,RocketMQ Namesrv 就启动完成。输出日志如下:
01:45:34.236 [NettyEventExecutor] INFO RocketmqRemoting - NettyEventExecutor service started
01:45:34.236 [FileWatchService] INFO RocketmqCommon - FileWatchService service started
最后,这是一个可选的步骤,命令行中输入 telnet 127.0.0.1 9876 ,看看是否能连接上 RocketMQ Namesrv 。
1.4.启动 RocketMQ Broker
打开 org.apache.rocketmq.broker.BrokerControllerTest 单元测试类,参考 #testBrokerRestart() 方法,我们编写 #main(String[] args) 方法,代码如下:
// BrokerControllerTest.java
public static void main(String[] args) throws Exception {
// 设置版本号
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
// NettyServerConfig 配置
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(10911);
// BrokerConfig 配置
final BrokerConfig brokerConfig = new BrokerConfig();
brokerConfig.setBrokerName("broker-a");
brokerConfig.setNamesrvAddr("127.0.0.1:9876");
// MessageStoreConfig 配置
final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setDeleteWhen("04");
messageStoreConfig.setFileReservedTime(48);
messageStoreConfig.setFlushDiskType(FlushDiskType.ASYNC_FLUSH);
messageStoreConfig.setDuplicationEnable(false);
// BrokerPathConfigHelper.setBrokerConfigPath("/Users/yunai/百度云同步盘/开发/Javascript/Story/incubator-rocketmq/conf/broker.conf");
// 创建 BrokerController 对象,并启动
BrokerController brokerController = new BrokerController(//
brokerConfig, //
nettyServerConfig, //
new NettyClientConfig(), //
messageStoreConfig);
brokerController.initialize();
brokerController.start();
// 睡觉,就不起来
System.out.println("你猜");
Thread.sleep(DateUtils.MILLIS_PER_DAY);
}
然后,右键运行,RocketMQ Broker 就启动完成了。输出日志如下:
你猜
不要懵逼,我们打开下 RocketMQ Namesrv 那,已经输出日志如下:
01:52:21.930 [RemotingExecutorThread_1] DEBUG RocketmqNamesrv - receive request, 103 127.0.0.1:62023 RemotingCommand [code=103, language=JAVA, version=293, opaque=0, flag(B)=0, remark=null, extFields={brokerId=0, bodyCrc32=137512002, clusterName=DefaultCluster, brokerAddr=192.0.108.1:10911, haServerAddr=192.0.108.1:10912, compressed=false, brokerName=broker-a}, serializeTypeCurrentRPC=JSON]
01:52:21.956 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, BenchmarkTest QueueData [brokerName=broker-a, readQueueNums=1024, writeQueueNums=1024, perm=6, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, OFFSET_MOVED_EVENT QueueData [brokerName=broker-a, readQueueNums=1, writeQueueNums=1, perm=6, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, broker-a QueueData [brokerName=broker-a, readQueueNums=1, writeQueueNums=1, perm=7, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, TBW102 QueueData [brokerName=broker-a, readQueueNums=8, writeQueueNums=8, perm=7, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, SELF_TEST_TOPIC QueueData [brokerName=broker-a, readQueueNums=1, writeQueueNums=1, perm=6, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new topic registered, DefaultCluster QueueData [brokerName=broker-a, readQueueNums=16, writeQueueNums=16, perm=7, topicSynFlag=0]
01:52:21.957 [RemotingExecutorThread_1] INFO RocketmqNamesrv - new broker registered, 192.0.108.1:10911 HAServer: 192.0.108.1:10912
妥妥的,原来 RocketMQ Broker 已经启动完成,并且注册到 RocketMQ Namesrv 上。
最后,这是一个可选的步骤,命令行中输入 telnet 127.0.0.1 10911 ,看看是否能连接上 RocketMQ Broker 。
1.5.启动 RocketMQ Producer
打开 org.apache.rocketmq.example.quickstart.Producer 示例类,代码如下:
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.example.quickstart;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
/**
* This class demonstrates how to send messages to brokers using provided {@link DefaultMQProducer}.
*/
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
/*
* Instantiate with a producer group name.
*/
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
/*
* Specify name server addresses.
* <p/>
*
* Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
* <pre>
* {@code
* producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
* }
* </pre>
*/
/*
* Launch the instance.
*/
producer.setNamesrvAddr("127.0.0.1:9876"); // <x> 哈哈哈哈
producer.start();
for (int i = 0; i < 1000; i++) {
try {
/*
* Create a message instance, specifying topic, tag and message body.
*/
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
/*
* Call send message to deliver message to one of brokers.
*/
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
/*
* Shut down once the producer instance is not longer in use.
*/
producer.shutdown();
}
}
注意,在 <x> 哈哈哈哈处,我们增加了 producer.setNamesrvAddr("127.0.0.1:9876") 代码块,指明 Producer 使用的 RocketMQ Namesrv 。
然后,右键运行,RocketMQ Producer 就启动完成。输出日志如下:
1.6.启动 RocketMQ Consumer
打开 org.apache.rocketmq.example.quickstart.Consumer 示例类,代码如下:
// Consumer.java
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
/*
* Instantiate with specified consumer group name.
*/
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
/*
* Specify name server addresses.
* <p/>
*
* Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
* <pre>
* {@code
* consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
* }
* </pre>
*/
/*
* Specify where to start in case the specified consumer group is a brand new one.
*/
consumer.setNamesrvAddr("127.0.0.1:9876"); // <x> 哈哈哈哈
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
/*
* Subscribe one more more topics to consume.
*/
consumer.subscribe("TopicTest", "*");
/*
* Register callback to execute on arrival of messages fetched from brokers.
*/
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
/*
* Launch the consumer instance.
*/
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
注意,在 <x> 哈哈哈哈处,我们还增加了 consumer.setNamesrvAddr("127.0.0.1:9876") 代码块,指明 Consumer 使用的 RocketMQ Namesrv 。
然后,右键运行,RocketMQ Consumer 就启动完成。输入日志如下:
网友评论