rocketMQ是阿里開源的一款十分優(yōu)秀的消息隊列，rocketMQ具有很多其他消息隊列不具有的特性，像事務(wù)消息，推拉結(jié)合的Api等等。更重要的是rocketMQ是用java開發(fā)的，像RabbitMQ這種用erlang進行開發(fā)，對于我們搞java的來說學(xué)習(xí)成本不言而喻。從14年開始使用rocketMQ開始，就對rocketMQ有很大的興趣，對于消息隊列的實現(xiàn)機制也是十分的好奇，并且，rocketMQ更是加入了apache。這也進一步的激發(fā)了我的學(xué)習(xí)興趣，和幾個小伙伴商量開始學(xué)習(xí)rocketMQ的源碼。
目前RocketMQ的代碼托管在github上：
老的地址:https://github.com/alibaba/RocketMQ
新的地址:https://github.com/apache/incubator-rocketmq
由于阿里向apache捐贈了rocketMQ，所以下面的地址算是遷移過去的。
但是兩者之間是有差異的，Apache上的代碼注釋幾乎全被刪掉了，并且有一些啟動腳本也被刪掉了，兩者的代碼目前差別不大，我選擇的是后者，同時和前者對照著學(xué)習(xí)。

無論學(xué)什么，尤其是這種比較大的開源項目，那么首先就是通過一個使用場景進行切入。首先我們需要在本地搭建一個可以debug的開發(fā)環(huán)境，通過producer生產(chǎn)一個消息和consumer消費一個消息這樣的流程來學(xué)習(xí)rocketMQ，首先將rocketMQ fork一下，方便自己來注釋。

目前整體的代碼結(jié)構(gòu)是這樣的：

Paste_Image.png

可以看到rocketMQ的分包還是十分的清晰的，主要幾塊是broker、namesrv、client和store。remoting是一個通信層，只關(guān)注實現(xiàn)的話，可以暫且放一放remoting。

首先我們需要在我的idea下跑通namesrv和broker，同時可以利用producer和consumer進行收發(fā)消息。對于Apache版本的項目編譯打包可能有些困難，我們可以參考老的版本的install.sh進行打包：

Paste_Image.png

打包歸打包，但是我們暫時可能用不上，我們的目的是在idea上能將namesrv和broker跑起來，我們找到namesrv的啟動腳本mqnamesrv,這里主要調(diào)用了runserver腳本，namesrv的啟動入口是com.alibaba.rocketmq.namesrv.Namesrv
Startup類。runserver.sh里主要做了一些jdk參數(shù)的配置：

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} $@

這里主要設(shè)置了堆內(nèi)存的大小，垃圾回收方式和gc日志等。我們不使用腳本啟動，因為我們要調(diào)試，那么直接切到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進行設(shè)置，這樣就能保證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");

主要是設(shè)置了rocketMQhome和namesrv的地址。這樣broker就可以連接上namesrv。
接下來我們就要寫producer和namesrv了，其實生產(chǎn)者和消費者的例子是自帶的，在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();
    }
}

這里主要是循環(huán)想TopicTest發(fā)送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");
    }
}

這里主要是以監(jiān)聽的方式進行消費。

那么先啟動namesrv再啟動broker，然后開始啟動producer最后啟動consumer。我們可以看到消息可以正常的接收和發(fā)送了。