1，createBrokerController
System.setProperty(RemotingCommand.RemotingVersionKey, Integer.toString(MQVersion.CurrentVersion));
// Socket发送缓冲区大小
if (null == System.getProperty(NettySystemConfig.SystemPropertySocketSndbufSize)) {
NettySystemConfig.SocketSndbufSize = 131072;
}
默认发送缓冲区大小128M
// Socket接收缓冲区大小
if (null == System.getProperty(NettySystemConfig.SystemPropertySocketRcvbufSize)) {
NettySystemConfig.SocketRcvbufSize = 131072;
}
默认接收缓冲区大小128M
// 检测包冲突
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";
默认集群名称
@ImportantField
private long brokerId = MixAll.MASTER_ID;
master集群ID 必须为0
private int brokerPermission = PermName.PERM_READ | PermName.PERM_WRITE;
broker权限 默认读与写
private int defaultTopicQueueNums = 8;
默认队列数量
// 自动创建Topic功能是否开启（线上建议关闭）
@ImportantField
private boolean autoCreateTopicEnable = true;
// 自动创建以集群名字命名的Topic功能是否开启
private boolean clusterTopicEnable = true;
// 自动创建以服务器名字命名的Topic功能是否开启
private boolean brokerTopicEnable = true;
// 自动创建订阅组功能是否开启（线上建议关闭）
@ImportantField
private boolean autoCreateSubscriptionGroup = true;
private int sendMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 4;
发送消息线程池数量
private int pullMessageThreadPoolNums = 16 + Runtime.getRuntime().availableProcessors() * 2;
订阅消息线程池数量
private int adminBrokerThreadPoolNums = 16;
adminBroker 线程池数量 各自独立的线程池 做到线程池隔离
private int clientManageThreadPoolNums = 16;
clientManage线程池数量
private int flushConsumerOffsetInterval = 1000 * 5;
flush消费端 文件偏移时间戳 默认5秒
private int flushConsumerOffsetHistoryInterval = 1000 * 60;
flush消费端 历史文件偏移时间戳 默认1
// 是否拒绝接收事务消息
@ImportantField
private boolean rejectTransactionMessage = false;
// 是否从地址服务器寻找Name Server地址，正式发布后，默认值为false
@ImportantField
private boolean fetchNamesrvAddrByAddressServer = false;
// 发送消息对应的线程池阻塞队列size
private int sendThreadPoolQueueCapacity = 100000;
// 订阅消息对应的线程池阻塞队列size
private int pullThreadPoolQueueCapacity = 100000;
// 过滤服务器数量
private int filterServerNums = 0;
// Consumer订阅消息时，Broker是否开启长轮询
private boolean longPollingEnable = true;
// 如果是短轮询，服务器挂起时间
private long shortPollingTimeMills = 1000;
// notify consumerId changed 开关
private boolean notifyConsumerIdsChangedEnable = true;
// slave 是否需要纠正位点
private boolean offsetCheckInSlave = false;
3，final NettyServerConfig nettyServerConfig = new NettyServerConfig();
netty 服务配置
private int listenPort = 8888; 监听端口
private int serverWorkerThreads = 8;Netty服务工作线程数量
private int serverCallbackExecutorThreads = 0;Netty服务异步回调线程池线程数量
private int serverSelectorThreads = 3;Netty Selector线程数量
private int serverOnewaySemaphoreValue = 256;控制单向的信号量 一次请求没有响应
private int serverAsyncSemaphoreValue = 64;控制异步信号量
private int serverChannelMaxIdleTimeSeconds = 120;服务空闲心跳检测时间间隔 单位秒
private int serverSocketSndBufSize = NettySystemConfig.SocketSndbufSize;
Netty发送缓冲区大小
private int serverSocketRcvBufSize = NettySystemConfig.SocketRcvbufSize;
Netty接受缓冲区大小
private boolean serverPooledByteBufAllocatorEnable = false;是否使用Netty内存池
4，final NettyClientConfig nettyClientConfig = new NettyClientConfig();
private int clientWorkerThreads = 4;Netty客户端工作线程
private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
Netty服务异步回调线程池线程数量
private int clientOnewaySemaphoreValue = NettySystemConfig.ClientOnewaySemaphoreValue;
控制单向的信号量 一次请求没有响应
private int clientAsyncSemaphoreValue = NettySystemConfig.ClientAsyncSemaphoreValue;
控制异步信号量
private long connectTimeoutMillis = 3000;
连接超时时间
private long channelNotActiveInterval = 1000 * 60;
扫描channel未激活时间戳
private int clientChannelMaxIdleTimeSeconds = 120;
客户端闲心跳检测时间间隔 单位秒
private int clientSocketSndBufSize = NettySystemConfig.SocketSndbufSize;
客户端发送缓冲区大小
private int clientSocketRcvBufSize = NettySystemConfig.SocketRcvbufSize;
客户端接受缓冲区大小
private boolean clientPooledByteBufAllocatorEnable = false;
客户端是否支持netty内存池
nettyServerConfig.setListenPort(10911);
监听端口
5，final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
// 存储跟目录
@ImportantField
private String storePathRootDir = System.getProperty("user.home") + File.separator + "store";
// CommitLog存储目录
@ImportantField
private String storePathCommitLog = System.getProperty("user.home") + File.separator + "store"
+ File.separator + "commitlog";
// CommitLog每个文件大小 1G
private int mapedFileSizeCommitLog = 1024 * 1024 * 1024;
// ConsumeQueue每个文件大小 默认存储30W条消息
private int mapedFileSizeConsumeQueue = 300000 * ConsumeQueue.CQStoreUnitSize;
// CommitLog刷盘间隔时间（单位毫秒）
@ImportantField
private int flushIntervalCommitLog = 1000;
// 是否定时方式刷盘，默认是实时刷盘
@ImportantField
private boolean flushCommitLogTimed = false;
// ConsumeQueue刷盘间隔时间（单位毫秒）
private int flushIntervalConsumeQueue = 1000;
// 清理资源间隔时间（单位毫秒）
private int cleanResourceInterval = 10000;
// 删除多个CommitLog文件的间隔时间（单位毫秒）
private int deleteCommitLogFilesInterval = 100;
// 删除多个ConsumeQueue文件的间隔时间（单位毫秒）
private int deleteConsumeQueueFilesInterval = 100;
// 强制删除文件间隔时间（单位毫秒）
private int destroyMapedFileIntervalForcibly = 1000 * 120;
// 定期检查Hanged文件间隔时间（单位毫秒）
private int redeleteHangedFileInterval = 1000 * 120;
// 何时触发删除文件, 默认凌晨4点删除文件
@ImportantField
private String deleteWhen = "04";
// 磁盘空间最大使用率
private int diskMaxUsedSpaceRatio = 75;
// 文件保留时间（单位小时）
@ImportantField
private int fileReservedTime = 72;
// 写消息索引到ConsumeQueue，缓冲区高水位，超过则开始流控
private int putMsgIndexHightWater = 600000;
// 最大消息大小，默认512K
private int maxMessageSize = 1024 * 512;
// 重启时，是否校验CRC
private boolean checkCRCOnRecover = true;
// 刷CommitLog，至少刷几个PAGE
private int flushCommitLogLeastPages = 4;
// 刷ConsumeQueue，至少刷几个PAGE
private int flushConsumeQueueLeastPages = 2;
// 刷CommitLog，彻底刷盘间隔时间
private int flushCommitLogThoroughInterval = 1000 * 10;
// 刷ConsumeQueue，彻底刷盘间隔时间
private int flushConsumeQueueThoroughInterval = 1000 * 60;
// 最大被拉取的消息字节数，消息在内存
@ImportantField
private int maxTransferBytesOnMessageInMemory = 1024 * 256;
// 最大被拉取的消息个数，消息在内存
@ImportantField
private int maxTransferCountOnMessageInMemory = 32;
// 最大被拉取的消息字节数，消息在磁盘
@ImportantField
private int maxTransferBytesOnMessageInDisk = 1024 * 64;
// 最大被拉取的消息个数，消息在磁盘
@ImportantField
private int maxTransferCountOnMessageInDisk = 8;
// 命中消息在内存的最大比例
@ImportantField
private int accessMessageInMemoryMaxRatio = 40;
// 是否开启消息索引功能
@ImportantField
private boolean messageIndexEnable = true;
private int maxHashSlotNum = 5000000;
private int maxIndexNum = 5000000 * 4;
private int maxMsgsNumBatch = 64;
// 是否使用安全的消息索引功能，即可靠模式。
// 可靠模式下，异常宕机恢复慢
// 非可靠模式下，异常宕机恢复快
@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;
// 如果不设置，则从NameServer获取Master HA服务地址
@ImportantField
private String haMasterAddress = null;
// Slave落后Master超过此值，则认为存在异常
private int haSlaveFallbehindMax = 1024 * 1024 * 256;
@ImportantField
private BrokerRole brokerRole = BrokerRole.ASYNC_MASTER;
@ImportantField
private FlushDiskType flushDiskType = FlushDiskType.ASYNC_FLUSH;
// 同步刷盘超时时间
private int syncFlushTimeout = 1000 * 5;
// 定时消息相关
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;
// 磁盘空间超过90%警戒水位，自动开始删除文件
@ImportantField
private boolean cleanFileForciblyEnable = true;
6,// 如果是slave，修改默认值
if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
}
如果是master broker命中消息在内存的最大比例是40
如果是slave broker命中消息在内存的最大比例是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);
}
}
检测Name Server地址设置是否正确 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监听Slave请求的端口，默认为服务端口+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);
初始化服务控制对象
this.brokerConfig = brokerConfig; broker配置
this.nettyServerConfig = nettyServerConfig;netty服务配置
this.nettyClientConfig = nettyClientConfig;netty客户端配置
this.messageStoreConfig = messageStoreConfig;消息存储配置
this.consumerOffsetManager = new ConsumerOffsetManager(this);
消费进度管理
this.topicConfigManager = new TopicConfigManager(this);
Topic配置管理
this.pullMessageProcessor = new PullMessageProcessor(this);
拉消息请求处理
this.pullRequestHoldService = new PullRequestHoldService(this);
拉消息请求管理，如果拉不到消息，则在这里Hold住，等待消息到来
this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
ConsumerId列表变化，通知所有Consumer
this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
Consumer连接、订阅关系管理
this.producerManager = new ProducerManager();
管理Producer组及各个Producer连接
this.clientHousekeepingService = new ClientHousekeepingService(this);
定期检测客户端连接，清除不活动的连接
this.broker2Client = new Broker2Client(this);
Broker主动调用客户端接口
this.subscriptionGroupManager = new SubscriptionGroupManager(this);
用来管理订阅组，包括订阅权限等
this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
Broker对外调用的API封装
this.filterServerManager = new FilterServerManager(this);
过滤服务管理
if (this.brokerConfig.getNamesrvAddr() != null) {
this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
log.info("user specfied name server address: {}",this.brokerConfig.getNamesrvAddr());
}
更新namesrv地址，防止存在存留数据
this.slaveSynchronize = new SlaveSynchronize(this);
Slave从Master同步信息（非消息）
this.sendThreadPoolQueue =
new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
初始化发送消息对应的线程池阻塞队列size
this.pullThreadPoolQueue =
new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());
初始化订阅消息对应的线程池阻塞队列size
this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());
broker 状态管理
this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this
.getNettyServerConfig().getListenPort()));
暂时没发现在哪用，等用到了在说
