1限寞，createBrokerController
System.setProperty(RemotingCommand.RemotingVersionKey, Integer.toString(MQVersion.CurrentVersion));
// Socket發(fā)送緩沖區(qū)大小
if (null == System.getProperty(NettySystemConfig.SystemPropertySocketSndbufSize)) {
NettySystemConfig.SocketSndbufSize = 131072;
}
默認(rèn)發(fā)送緩沖區(qū)大小128M
// Socket接收緩沖區(qū)大小
if (null == System.getProperty(NettySystemConfig.SystemPropertySocketRcvbufSize)) {
NettySystemConfig.SocketRcvbufSize = 131072;
}
默認(rèn)接收緩沖區(qū)大小128M
// 檢測(cè)包沖突
PackageConflictDetect.detectFastjson();
// 解析命令行
Options options = ServerUtil.buildCommandlineOptions(new Options());
commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
new PosixParser());
if (null == commandLine) {
System.exit(-1);
return null;
}
2酷愧，final BrokerConfig brokerConfig = new BrokerConfig();
private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
System.getenv(MixAll.ROCKETMQ_HOME_ENV));
獲取rocketMq目錄
@ImportantField
private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY,
System.getenv(MixAll.NAMESRV_ADDR_ENV));
獲取namesrc地址
@ImportantField
private String brokerIP1 = RemotingUtil.getLocalAddress();
獲取本地地址
private String brokerIP2 = RemotingUtil.getLocalAddress();
獲取本地地址
@ImportantField
private String brokerName = localHostName();
獲取broker hostname
@ImportantField
private String brokerClusterName = "DefaultCluster";
默認(rèn)集群名稱
@ImportantField
private long brokerId = MixAll.MASTER_ID;
master集群ID 必須為0
private int brokerPermission = PermName.PERM_READ | PermName.PERM_WRITE;
broker權(quán)限 默認(rèn)讀與寫(xiě)
private int defaultTopicQueueNums = 8;
默認(rèn)隊(duì)列數(shù)量
// 自動(dòng)創(chuàng)建Topic功能是否開(kāi)啟（線上建議關(guān)閉）
@ImportantField
private boolean autoCreateTopicEnable = true;
// 自動(dòng)創(chuàng)建以集群名字命名的Topic功能是否開(kāi)啟
private boolean clusterTopicEnable = true;
// 自動(dòng)創(chuàng)建以服務(wù)器名字命名的Topic功能是否開(kāi)啟
private boolean brokerTopicEnable = true;
// 自動(dòng)創(chuàng)建訂閱組功能是否開(kāi)啟（線上建議關(guān)閉）
@ImportantField
private boolean autoCreateSubscriptionGroup = true;
private int sendMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 4;
發(fā)送消息線程池?cái)?shù)量
private int pullMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
訂閱消息線程池?cái)?shù)量
private int adminBrokerThreadPoolNums = 16;
adminBroker 線程池?cái)?shù)量 各自獨(dú)立的線程池 做到線程池隔離
private int clientManageThreadPoolNums = 16;
clientManage線程池?cái)?shù)量
private int flushConsumerOffsetInterval = 1000 * 5;
flush消費(fèi)端 文件偏移時(shí)間戳 默認(rèn)5秒
private int flushConsumerOffsetHistoryInterval = 1000 * 60;
flush消費(fèi)端 歷史文件偏移時(shí)間戳 默認(rèn)1
// 是否拒絕接收事務(wù)消息
@ImportantField
private boolean rejectTransactionMessage = false;
// 是否從地址服務(wù)器尋找Name Server地址，正式發(fā)布后广鳍，默認(rèn)值為false
@ImportantField
private boolean fetchNamesrvAddrByAddressServer = false;
// 發(fā)送消息對(duì)應(yīng)的線程池阻塞隊(duì)列size
private int sendThreadPoolQueueCapacity = 100000;
// 訂閱消息對(duì)應(yīng)的線程池阻塞隊(duì)列size
private int pullThreadPoolQueueCapacity = 100000;
// 過(guò)濾服務(wù)器數(shù)量
private int filterServerNums = 0;
// Consumer訂閱消息時(shí)，Broker是否開(kāi)啟長(zhǎng)輪詢
private boolean longPollingEnable = true;
// 如果是短輪詢姑蓝，服務(wù)器掛起時(shí)間
private long shortPollingTimeMills = 1000;
// notify consumerId changed 開(kāi)關(guān)
private boolean notifyConsumerIdsChangedEnable = true;
// slave 是否需要糾正位點(diǎn)
private boolean offsetCheckInSlave = false;
3膝蜈，final NettyServerConfig nettyServerConfig = new NettyServerConfig();
netty 服務(wù)配置
private int listenPort = 8888; 監(jiān)聽(tīng)端口
private int serverWorkerThreads = 8;Netty服務(wù)工作線程數(shù)量
private int serverCallbackExecutorThreads = 0;Netty服務(wù)異步回調(diào)線程池線程數(shù)量
private int serverSelectorThreads = 3;Netty Selector線程數(shù)量
private int serverOnewaySemaphoreValue = 256;控制單向的信號(hào)量 一次請(qǐng)求沒(méi)有響應(yīng)
private int serverAsyncSemaphoreValue = 64;控制異步信號(hào)量
private int serverChannelMaxIdleTimeSeconds = 120;服務(wù)空閑心跳檢測(cè)時(shí)間間隔 單位秒
private int serverSocketSndBufSize = NettySystemConfig.SocketSndbufSize;
Netty發(fā)送緩沖區(qū)大小
private int serverSocketRcvBufSize = NettySystemConfig.SocketRcvbufSize;
Netty接受緩沖區(qū)大小
private boolean serverPooledByteBufAllocatorEnable = false;是否使用Netty內(nèi)存池
4锅移，final NettyClientConfig nettyClientConfig = new NettyClientConfig();
private int clientWorkerThreads = 4;Netty客戶端工作線程
private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
Netty服務(wù)異步回調(diào)線程池線程數(shù)量
private int clientOnewaySemaphoreValue = NettySystemConfig.ClientOnewaySemaphoreValue;
控制單向的信號(hào)量 一次請(qǐng)求沒(méi)有響應(yīng)
private int clientAsyncSemaphoreValue = NettySystemConfig.ClientAsyncSemaphoreValue;
控制異步信號(hào)量
private long connectTimeoutMillis = 3000;
連接超時(shí)時(shí)間
private long channelNotActiveInterval = 1000 * 60;
掃描channel未激活時(shí)間戳
private int clientChannelMaxIdleTimeSeconds = 120;
客戶端閑心跳檢測(cè)時(shí)間間隔 單位秒
private int clientSocketSndBufSize = NettySystemConfig.SocketSndbufSize;
客戶端發(fā)送緩沖區(qū)大小
private int clientSocketRcvBufSize = NettySystemConfig.SocketRcvbufSize;
客戶端接受緩沖區(qū)大小
private boolean clientPooledByteBufAllocatorEnable = false;
客戶端是否支持netty內(nèi)存池
nettyServerConfig.setListenPort(10911);
監(jiān)聽(tīng)端口
5，final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
// 存儲(chǔ)跟目錄
@ImportantField
private String storePathRootDir = System.getProperty("user.home") + File.separator + "store";
// CommitLog存儲(chǔ)目錄
@ImportantField
private String storePathCommitLog = System.getProperty("user.home") + File.separator + "store"
+ File.separator + "commitlog";
// CommitLog每個(gè)文件大小 1G
private int mapedFileSizeCommitLog = 1024 * 1024 * 1024;
// ConsumeQueue每個(gè)文件大小 默認(rèn)存儲(chǔ)30W條消息
private int mapedFileSizeConsumeQueue = 300000 * ConsumeQueue.CQStoreUnitSize;
// CommitLog刷盤間隔時(shí)間（單位毫秒）
@ImportantField
private int flushIntervalCommitLog = 1000;
// 是否定時(shí)方式刷盤饱搏，默認(rèn)是實(shí)時(shí)刷盤
@ImportantField
private boolean flushCommitLogTimed = false;
// ConsumeQueue刷盤間隔時(shí)間（單位毫秒）
private int flushIntervalConsumeQueue = 1000;
// 清理資源間隔時(shí)間（單位毫秒）
private int cleanResourceInterval = 10000;
// 刪除多個(gè)CommitLog文件的間隔時(shí)間（單位毫秒）
private int deleteCommitLogFilesInterval = 100;
// 刪除多個(gè)ConsumeQueue文件的間隔時(shí)間（單位毫秒）
private int deleteConsumeQueueFilesInterval = 100;
// 強(qiáng)制刪除文件間隔時(shí)間（單位毫秒）
private int destroyMapedFileIntervalForcibly = 1000 * 120;
// 定期檢查Hanged文件間隔時(shí)間（單位毫秒）
private int redeleteHangedFileInterval = 1000 * 120;
// 何時(shí)觸發(fā)刪除文件, 默認(rèn)凌晨4點(diǎn)刪除文件
@ImportantField
private String deleteWhen = "04";
// 磁盤空間最大使用率
private int diskMaxUsedSpaceRatio = 75;
// 文件保留時(shí)間（單位小時(shí)）
@ImportantField
private int fileReservedTime = 72;
// 寫(xiě)消息索引到ConsumeQueue非剃，緩沖區(qū)高水位，超過(guò)則開(kāi)始流控
private int putMsgIndexHightWater = 600000;
// 最大消息大小推沸，默認(rèn)512K
private int maxMessageSize = 1024 * 512;
// 重啟時(shí)备绽，是否校驗(yàn)CRC
private boolean checkCRCOnRecover = true;
// 刷CommitLog，至少刷幾個(gè)PAGE
private int flushCommitLogLeastPages = 4;
// 刷ConsumeQueue鬓催，至少刷幾個(gè)PAGE
private int flushConsumeQueueLeastPages = 2;
// 刷CommitLog肺素，徹底刷盤間隔時(shí)間
private int flushCommitLogThoroughInterval = 1000 * 10;
// 刷ConsumeQueue，徹底刷盤間隔時(shí)間
private int flushConsumeQueueThoroughInterval = 1000 * 60;
// 最大被拉取的消息字節(jié)數(shù)深浮，消息在內(nèi)存
@ImportantField
private int maxTransferBytesOnMessageInMemory = 1024 * 256;
// 最大被拉取的消息個(gè)數(shù)压怠，消息在內(nèi)存
@ImportantField
private int maxTransferCountOnMessageInMemory = 32;
// 最大被拉取的消息字節(jié)數(shù)，消息在磁盤
@ImportantField
private int maxTransferBytesOnMessageInDisk = 1024 * 64;
// 最大被拉取的消息個(gè)數(shù)飞苇，消息在磁盤
@ImportantField
private int maxTransferCountOnMessageInDisk = 8;
// 命中消息在內(nèi)存的最大比例
@ImportantField
private int accessMessageInMemoryMaxRatio = 40;
// 是否開(kāi)啟消息索引功能
@ImportantField
private boolean messageIndexEnable = true;
private int maxHashSlotNum = 5000000;
private int maxIndexNum = 5000000 * 4;
private int maxMsgsNumBatch = 64;
// 是否使用安全的消息索引功能菌瘫，即可靠模式。
// 可靠模式下布卡，異常宕機(jī)恢復(fù)慢
// 非可靠模式下雨让，異常宕機(jī)恢復(fù)快
@ImportantField
private boolean messageIndexSafe = false;
// HA功能
private int haListenPort = 10912;
private int haSendHeartbeatInterval = 1000 * 5;
private int haHousekeepingInterval = 1000 * 20;
private int haTransferBatchSize = 1024 * 32;
// 如果不設(shè)置，則從NameServer獲取Master HA服務(wù)地址
@ImportantField
private String haMasterAddress = null;
// Slave落后Master超過(guò)此值忿等，則認(rèn)為存在異常
private int haSlaveFallbehindMax = 1024 * 1024 * 256;
@ImportantField
private BrokerRole brokerRole = BrokerRole.ASYNC_MASTER;
@ImportantField
private FlushDiskType flushDiskType = FlushDiskType.ASYNC_FLUSH;
// 同步刷盤超時(shí)時(shí)間
private int syncFlushTimeout = 1000 * 5;
// 定時(shí)消息相關(guān)
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
private long flushDelayOffsetInterval = 1000 * 10;
// 磁盤空間超過(guò)90%警戒水位栖忠，自動(dòng)開(kāi)始刪除文件
@ImportantField
private boolean cleanFileForciblyEnable = true;
6,// 如果是slave，修改默認(rèn)值
if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
}
如果是master broker命中消息在內(nèi)存的最大比例是40
如果是slave broker命中消息在內(nèi)存的最大比例是30
7,String namesrvAddr = brokerConfig.getNamesrvAddr();
if (null != namesrvAddr) {
try {
String[] addrArray = namesrvAddr.split(";");
if (addrArray != null) {
for (String addr : addrArray) {
RemotingUtil.string2SocketAddress(addr);
}
}
}
catch (Exception e) {
System.out
.printf(
"The Name Server Address[%s] illegal, please set it as follows, "127.0.0.1:9876;192.168.0.1:9876"\n",
namesrvAddr);
System.exit(-3);
}
}
檢測(cè)Name Server地址設(shè)置是否正確 IP:PORT
7, switch (messageStoreConfig.getBrokerRole()) {
case ASYNC_MASTER:
case SYNC_MASTER:
brokerConfig.setBrokerId(MixAll.MASTER_ID);
break;
case SLAVE:
if (brokerConfig.getBrokerId() <= 0) {
System.out.println("Slave's brokerId must be > 0");
System.exit(-3);
}
break;
default:
break;
}
判斷如果是Master Broker Id必須是0 ,如果是Slave Broker Id必須大于0
8,// Master監(jiān)聽(tīng)Slave請(qǐng)求的端口贸街，默認(rèn)為服務(wù)端口+1
messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
// 初始化Logback
LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
JoranConfigurator configurator = new JoranConfigurator();
configurator.setContext(lc);
lc.reset();
configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
log = LoggerFactory.getLogger(LoggerName.BrokerLoggerName);
9,final BrokerController controller = new BrokerController(
brokerConfig,
nettyServerConfig,
nettyClientConfig,
messageStoreConfig);
初始化服務(wù)控制對(duì)象
this.brokerConfig = brokerConfig; broker配置
this.nettyServerConfig = nettyServerConfig;netty服務(wù)配置
this.nettyClientConfig = nettyClientConfig;netty客戶端配置
this.messageStoreConfig = messageStoreConfig;消息存儲(chǔ)配置
this.consumerOffsetManager = new ConsumerOffsetManager(this);
消費(fèi)進(jìn)度管理
this.topicConfigManager = new TopicConfigManager(this);
Topic配置管理
this.pullMessageProcessor = new PullMessageProcessor(this);
拉消息請(qǐng)求處理
this.pullRequestHoldService = new PullRequestHoldService(this);
拉消息請(qǐng)求管理庵寞，如果拉不到消息，則在這里Hold住薛匪，等待消息到來(lái)
this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
ConsumerId列表變化捐川，通知所有Consumer
this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
Consumer連接、訂閱關(guān)系管理
this.producerManager = new ProducerManager();
管理Producer組及各個(gè)Producer連接
this.clientHousekeepingService = new ClientHousekeepingService(this);
定期檢測(cè)客戶端連接逸尖，清除不活動(dòng)的連接
this.broker2Client = new Broker2Client(this);
Broker主動(dòng)調(diào)用客戶端接口
this.subscriptionGroupManager = new SubscriptionGroupManager(this);
用來(lái)管理訂閱組古沥，包括訂閱權(quán)限等
this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
Broker對(duì)外調(diào)用的API封裝
this.filterServerManager = new FilterServerManager(this);
過(guò)濾服務(wù)管理
if (this.brokerConfig.getNamesrvAddr() != null) {
this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
log.info("user specfied name server address: {}",this.brokerConfig.getNamesrvAddr());
}
更新namesrv地址，防止存在存留數(shù)據(jù)
this.slaveSynchronize = new SlaveSynchronize(this);
Slave從Master同步信息（非消息）
this.sendThreadPoolQueue =
new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
初始化發(fā)送消息對(duì)應(yīng)的線程池阻塞隊(duì)列size
this.pullThreadPoolQueue =
new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());
初始化訂閱消息對(duì)應(yīng)的線程池阻塞隊(duì)列size
this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());
broker 狀態(tài)管理
this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this
.getNettyServerConfig().getListenPort()));
暫時(shí)沒(méi)發(fā)現(xiàn)在哪用娇跟，等用到了在說(shuō)
