美文网首页实时数据相关程序员
他处找到的对参数讲的最全的

他处找到的对参数讲的最全的

作者: 运维开发笔记 | 来源:发表于2017-11-22 17:37 被阅读20次

    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()));
    暂时没发现在哪用,等用到了在说

    相关文章

      网友评论

        本文标题:他处找到的对参数讲的最全的

        本文链接:https://www.haomeiwen.com/subject/oyjovxtx.html