全站最硬核 百万字强肝RocketMq源码 火热更新中~（七十三）

Posted 2022-02-13 源码看过吗

心跳执行器：

this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
    this.brokerConfig.getHeartbeatThreadPoolNums(),
    this.brokerConfig.getHeartbeatThreadPoolNums(),
    1000 * 60,
    TimeUnit.MILLISECONDS,
    this.heartbeatThreadPoolQueue,
    new ThreadFactoryImpl("HeartbeatThread_", true));

事务执行器：

this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
    this.brokerConfig.getEndTransactionThreadPoolNums(),
    this.brokerConfig.getEndTransactionThreadPoolNums(),
    1000 * 60,
    TimeUnit.MILLISECONDS,
    this.endTransactionThreadPoolQueue,
    new ThreadFactoryImpl("EndTransactionThread_"));

消费者管理执行器：

this.consumerManageExecutor =
    Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
        "ConsumerManageThread_"));

注册一系列的执行器，其实就是为了合理安排线程

合理线程分工，达到提升性能的目的

注册处理器

this.registerProcessor();

这里我们先不细看

继续往下：

接下来又是一系列的定时任务：

开启broker状态的记录

final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
final long period = 1000 * 60 * 60 * 24;
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() 
    @Override
    public void run() 
        try 
            BrokerController.this.getBrokerStats().record();
         catch (Throwable e) 
            log.error("schedule record error.", e);
        
    
, initialDelay, period, TimeUnit.MILLISECONDS);