RocketMQ源码阅读——debug环境构建
rocketMQ是阿里开源的一款十分优秀的消息队列，rocketMQ具有很多其他消息队列不具有的特性，像事务消息，推拉结合的Api等等。更重要的是rocketMQ是用java开发的，像RabbitMQ这种用erlang进行开发，对于我们搞java的来说学习成本不言而喻。从14年开始使用rocketMQ开始，就对rocketMQ有很大的兴趣，对于消息队列的实现机制也是十分的好奇，并且，rocketMQ更是加入了apache。这也进一步的激发了我的学习兴趣，和几个小伙伴商量开始学习rocketMQ的源码。
目前RocketMQ的代码托管在github上：
老的地址:https://github.com/alibaba/RocketMQ
新的地址:https://github.com/apache/incubator-rocketmq
由于阿里向apache捐赠了rocketMQ，所以下面的地址算是迁移过去的。
但是两者之间是有差异的，Apache上的代码注释几乎全被删掉了，并且有一些启动脚本也被删掉了，两者的代码目前差别不大，我选择的是后者，同时和前者对照着学习。
无论学什么，尤其是这种比较大的开源项目，那么首先就是通过一个使用场景进行切入。首先我们需要在本地搭建一个可以debug的开发环境，通过producer生产一个消息和consumer消费一个消息这样的流程来学习rocketMQ，首先将rocketMQ fork一下，方便自己来注释。
目前整体的代码结构是这样的：
Paste_Image.png
可以看到rocketMQ的分包还是十分的清晰的，主要几块是broker、namesrv、client和store。remoting是一个通信层，只关注实现的话，可以暂且放一放remoting。
首先我们需要在我的idea下跑通namesrv和broker，同时可以利用producer和consumer进行收发消息。对于Apache版本的项目编译打包可能有些困难，我们可以参考老的版本的install.sh进行打包：
Paste_Image.png
打包归打包，但是我们暂时可能用不上，我们的目的是在idea上能将namesrv和broker跑起来，我们找到namesrv的启动脚本mqnamesrv,这里主要调用了runserver脚本，namesrv的启动入口是com.alibaba.rocketmq.namesrv.Namesrv
Startup类。runserver.sh里主要做了一些jdk参数的配置：
error_exit ()
{
echo "ERROR: $1 !!"
exit 1
}
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/opt/taobao/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"
export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
#===========================================================================================
# JVM Configuration
#===========================================================================================
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m"
JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:+DisableExplicitGC"
JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${HOME}/rmq_srv_gc.log -XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib"
#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
$JAVA ${JAVA_OPT} $@
这里主要设置了堆内存的大小，垃圾回收方式和gc日志等。我们不使用脚本启动，因为我们要调试，那么直接切到NamesrvStartup类去启动。
如果我们直接运行的话会报一个rocketMQ home not set 的错误，那么我们这里需要稍微改一下代码：
final NamesrvConfig namesrvConfig = new NamesrvConfig();
namesrvConfig.setRocketmqHome("/home/zhiming/git/incubator-rocketmq");
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(9876);
这里主要做的是手动的将namesrv的端口号和rocketMQhome进行设置，这样就能保证rocketMQ的namesrv能正常启动了。
接下来启动broker，同样的方式我们找到broker的启动类：com.alibaba.rocketmq.broker.BrokerStartup。同样我们也需要做一些改动：
final BrokerConfig brokerConfig = new BrokerConfig();
brokerConfig.setRocketmqHome("/home/zhiming/git/incubator-rocketmq");
brokerConfig.setNamesrvAddr("localhost:9876");
主要是设置了rocketMQhome和namesrv的地址。这样broker就可以连接上namesrv。
接下来我们就要写producer和namesrv了，其实生产者和消费者的例子是自带的，在example包中的com.alibaba.rocketmq.example.quickstart下，有一个Consumer和Producer：
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 com.alibaba.rocketmq.example.quickstart;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
producer.setNamesrvAddr("localhost:9876");
for (int i = 0; i < 1000; i++) {
try {
Message msg = new Message("TopicTest",
"TagA",
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
SendResult sendResult = producer.send(msg);
LocalTransactionExecuter tranExecuter = new LocalTransactionExecuter() {
@Override
public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
return null;
}
};
System.out.printf("%s%n", sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
}
这里主要是循环想TopicTest发送1000次消息。
Consumer:
/**
* 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 com.alibaba.rocketmq.example.quickstart;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
这里主要是以监听的方式进行消费。
那么先启动namesrv再启动broker，然后开始启动producer最后启动consumer。我们可以看到消息可以正常的接收和发送了。
