說在前面

本次開始進行rocketmq原始碼解析，比較喜歡rocketmq的架構設計，rocketmq內嵌了namesrv註冊中心儲存了元資料，進行負載均衡、容錯的一些處理，4.3以上支援訊息事務，有管理控制檯、命令列工具，底層namesrv與broker、client與server互動netty實現。更多精彩文章請關注“天河聊架構”微信公眾號。

 

原始碼解析

建立NamesrvController，進入這個方法org.apache.rocketmq.namesrv.NamesrvStartup#main，再進入這個方法org.apache.rocketmq.namesrv.NamesrvStartup#main0

public static NamesrvController main0(String[] args) {

        try {
//            原始碼解析之建立namesrv控制器 =》
            NamesrvController controller = createNamesrvController(args);
//            原始碼解析之啟動namesrv控制器 =》
            start(controller);
            String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
            log.info(tip);
            System.out.printf("%s%n", tip);
            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }

        return null;
    }

進入這個方法org.apache.rocketmq.namesrv.NamesrvStartup#createNamesrvController

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
//        從系統檔案中查詢rocketmq版本，預設4.3.0
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
//        PackageConflictDetect.detectFastjson();
//        構建命令列操作的指令 =》
        Options options = ServerUtil.buildCommandlineOptions(new Options());
//        mqnamesrv 啟動namesrv命令 =》
        commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
        if (null == commandLine) {
//            系統非正常退出，流程結束
            System.exit(-1);
            return null;
        }

//        解析配置檔案 =》
        final NamesrvConfig namesrvConfig = new NamesrvConfig();
//        =》
        final NettyServerConfig nettyServerConfig = new NettyServerConfig();
//        設定 namesrv的服務埠
        nettyServerConfig.setListenPort(9876);
//        c 指定啟動的時候載入配置檔案
        if (commandLine.hasOption('c')) {
//            命令列啟動指定配置檔案，前面用c開頭
            String file = commandLine.getOptionValue('c');
            if (file != null) {
                InputStream in = new BufferedInputStream(new FileInputStream(file));
                properties = new Properties();
                properties.load(in);
                MixAll.properties2Object(properties, namesrvConfig);
                MixAll.properties2Object(properties, nettyServerConfig);
//                設定命令列啟動namesrv指定的配置檔案路徑
                namesrvConfig.setConfigStorePath(file);
                System.out.printf("load config properties file OK, %s%n", file);
                in.close();
            }
        }

//        列印namesrv的配置資訊，命令列上面加p
        if (commandLine.hasOption('p')) {
            MixAll.printObjectProperties(null, namesrvConfig);
            MixAll.printObjectProperties(null, nettyServerConfig);
//            正常程式退出
            System.exit(0);
        }

//        把命令列屬性解析成properties
        MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
//        解析 ROCKETMQ_HOME 環境變數
        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);
        }

        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
        JoranConfigurator configurator = new JoranConfigurator();
        configurator.setContext(lc);
        lc.reset();
//        logback日誌檔案路徑
        configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
        log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
        MixAll.printObjectProperties(log, namesrvConfig);
        MixAll.printObjectProperties(log, nettyServerConfig);
//        建立namesrv控制器 =》
        final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
        // remember all configs to prevent discard 把配置檔案配置值的屬性值註冊到namesrv控制器
        controller.getConfiguration().registerConfig(properties);
        return controller;
    }

進入這個方法org.apache.rocketmq.srvutil.ServerUtil#buildCommandlineOptions 這裡可以看到啟動rocketmq的時候namesrv引數怎麼制定

public static Options buildCommandlineOptions(final Options options) {
        Option opt = new Option("h", "help", false, "Print help");
        opt.setRequired(false);
        options.addOption(opt);
//        這裡可以看到啟動的時候有一個n引數指定namesrv的地址，可以是單機可以是叢集，啟動地址之間用;分開
        opt =
            new Option("n", "namesrvAddr", true,
                "Name server address list, eg: 192.168.0.1:9876;192.168.0.2:9876");
        opt.setRequired(false);
        options.addOption(opt);
        return options;
    }

進入這個方法org.apache.rocketmq.namesrv.NamesrvStartup#buildCommandlineOptions 從這裡看出來啟動rocketmq的時候怎麼指定配置檔案，是否自動建立topic、消費組這些引數都可以在配置檔案中配置

//    指定namesrv啟動的時候載入的配置檔案
    public static Options buildCommandlineOptions(final Options options) {
        Option opt = new Option("c", "configFile", true, "Name server config properties file");
        opt.setRequired(false);
        options.addOption(opt);
//        控制檯輸出配置項
        opt = new Option("p", "printConfigItem", false, "Print all config item");
        opt.setRequired(false);
        options.addOption(opt);
        return options;
    }

從這裡可以看到一些配置的儲存地址

public class NamesrvConfig {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
//    解析ROCKETMQ_HOME
    private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
//    kvConfig儲存地址
    private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
//    namesrv儲存地址
    private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";

nettyserver的一些配置

public class NettyServerConfig implements Cloneable {
//    netty server監聽埠
    private int listenPort = 8888;
//    服務端工作執行緒數
    private int serverWorkerThreads = 8;
//    服務端回撥執行執行緒
    private int serverCallbackExecutorThreads = 0;
//    服務端選擇器執行緒
    private int serverSelectorThreads = 3;
//    oneway方式訊號量的值
    private int serverOnewaySemaphoreValue = 256;
//    服務端非同步訊號量值
    private int serverAsyncSemaphoreValue = 64;
//    服務端channel最大空閒時間
    private int serverChannelMaxIdleTimeSeconds = 120;
//    傳送訊息最大大小從系統屬性中獲取，預設值65535
    private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
//    接受訊息最大大小從系統屬性中獲取，預設值65535
    private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
    private boolean serverPooledByteBufAllocatorEnable = true;

nettyserver預設監聽埠9876

nettyServerConfig.setListenPort(9876);

從這裡可以看到是解析啟動的時候指定的配置檔案屬性

//        c 指定啟動的時候載入配置檔案
        if (commandLine.hasOption('c')) {
//            命令列啟動指定配置檔案，前面用c開頭
            String file = commandLine.getOptionValue('c');
            if (file != null) {
                InputStream in = new BufferedInputStream(new FileInputStream(file));
                properties = new Properties();
                properties.load(in);
                MixAll.properties2Object(properties, namesrvConfig);
                MixAll.properties2Object(properties, nettyServerConfig);

配置檔案中可以配置這個類org.apache.rocketmq.common.namesrv.NamesrvConfig namesrv的一些配置

//    解析ROCKETMQ_HOME
    private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
//    kvConfig儲存地址
    private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
//    namesrv儲存地址
    private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
    private String productEnvName = "center";
    private boolean clusterTest = false;
    private boolean orderMessageEnable = false;

也可以配置這個類org.apache.rocketmq.remoting.netty.NettyServerConfig nettyserver的一些配置

public class NettyServerConfig implements Cloneable {
//    netty server監聽埠
    private int listenPort = 8888;
//    服務端工作執行緒數
    private int serverWorkerThreads = 8;
//    服務端回撥執行執行緒
    private int serverCallbackExecutorThreads = 0;
//    服務端選擇器執行緒
    private int serverSelectorThreads = 3;
//    oneway方式訊號量的值
    private int serverOnewaySemaphoreValue = 256;
//    服務端非同步訊號量值
    private int serverAsyncSemaphoreValue = 64;
//    服務端channel最大空閒時間
    private int serverChannelMaxIdleTimeSeconds = 120;
//    傳送訊息最大大小從系統屬性中獲取，預設值65535
    private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
//    接受訊息最大大小從系統屬性中獲取，預設值65535
    private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
    private boolean serverPooledByteBufAllocatorEnable = true;

namesrv配置的儲存地址就是啟動指定的配置檔案

namesrvConfig.setConfigStorePath(file);

也可以在命令列上指定namesrv屬性配置

MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

logback的配置檔案

configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");

根據namesrv配置和nettyserver配置建立NamesrvController

final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

進入這個方法org.apache.rocketmq.namesrv.NamesrvController#NamesrvController

public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
        this.namesrvConfig = namesrvConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.kvConfigManager = new KVConfigManager(this);
        this.routeInfoManager = new RouteInfoManager();
        this.brokerHousekeepingService = new BrokerHousekeepingService(this);
//        =》
        this.configuration = new Configuration(
            log,
            this.namesrvConfig, this.nettyServerConfig
        );
//        指定儲存namesrv配置的配置檔案
        this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
    }

進入這裡org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#RouteInfoManager

public RouteInfoManager() {
//        指定長度，減少擴容提高效能
        this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);//topicQueue快取資訊
        this.brokerAddrTable = new HashMap<String, BrokerData>(128);//broker地址快取
        this.clusterAddrTable = new HashMap<String, Set<String>>(32);//叢集地址快取
        this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);//可用的broker快取
        this.filterServerTable = new HashMap<String, List<String>>(256);//過濾的server地址
    }

設定namrsrv配置儲存地址System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties"

this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");

// remember all configs to prevent discard 把配置檔案配置值的屬性值註冊到namesrv控制器
controller.getConfiguration().registerConfig(properties);

進入這個方法org.apache.rocketmq.common.Configuration#registerConfig(java.util.Properties)

public Configuration registerConfig(Object configObject) {
    try {
        readWriteLock.writeLock().lockInterruptibly();
        try {

            Properties registerProps = MixAll.object2Properties(configObject);
            merge(registerProps, this.allConfigs);
            configObjectList.add(configObject);
        } finally {
            readWriteLock.writeLock().unlock();
        }
    } catch (InterruptedException e) {
        log.error("registerConfig lock error");
    }
    return this;
}

/**
 * All properties include configs in object and extend properties.
 * 所有的啟動配置都在這裡
 */
private Properties allConfigs = new Properties();

 

說在最後

本次解析僅代表個人觀點，僅供參考。

 

加入技術微信群

相關推薦