broker是消息队列的核心组建,承载了消息接收,存储和转发的职责. 因此, broker需要具备各种基本功能和高阶功能.
1.基本功能

• 承载消息堆积的能力
  消息到达服务端如果不经过任何处理就到接收者了, broker就失去了它的意义. 为了满足我们错峰/流控/最终可达等一系列需求, 把消息存储下来, 然后选择时机投递就显得是顺理成章的了. 因此, broker必须具备强大的消息堆积能力.
• 消费关系解偶
  其实就是具备单播(点对点)和广播(一点对多点)功能.
  2.高阶功能
• 消息去重
  当前的消息队列还无法做到消息完全去重(除非允许消息丢失), 只能尽量减少消息重复的概率.
• 顺序消息
  消息有序指的是一类消息消费时, 能按照发送的顺序来消费. 例如: 一个订单产生了3条消息, 分别是订单创建, 订单付款, 订单完成. 消费时, 要按照这个顺序消费才能有意义. 但是同时订单之间是可以并行消费的. RocketMQ可以严格的保证消息有序.

我们从三个方面分析broker, 分别为broker的启动, 接收消息和消费消息.

broker的启动

broker的启动过程是一个比较复杂的过程, 其中涉及到很多broker模块自身的初始化, 例如：路由信息设置, 消息落地, 消息推送, 以及和namesvr模块通信部分的初始化.

broker启动可以分为两个步骤, initialize和start.

initialize

1.当broker启动命令得到外部传入的初始化参数以后, 将参数注入对应的config类当中, 这些config类包括:

• broker自身的配置：包括根目录, namesrv地址, broker的IP和名称, 消息队列数, 收发消息线程池数等参数
• netty启动配置：包括netty监听端口, 工作线程数, 异步发送消息信号量数量等网络配置等参数.
• 存储层配置：包括存储跟目录, CommitLog配置, 持久化策略配置等参数.
  这一步骤的具体代码在BrokerStartup的start方法和createBrokerController方法中, 较为简单,此处不贴出.

2.当配置信息设置完毕后, broker会将这些参数传入borkerController控制器当中, 这个控制器会初始加载很多的管理器, 如下:

• topicManager：用于管理broker中存储的所有topic的配置
• consumerOffsetManager：管理Consumer的消费进度
• subscriptionGroupManager：用来管理订阅组, 包括订阅权限等
• messageStore：用于broker层的消息落地存储.

这部分代码在BrokerController.initialize中, 代码片段如下:

public boolean initialize() throws CloneNotSupportedException {
    boolean result = true;
    //加载topicConfigManager
    result = result && this.topicConfigManager.load();
    //加载consumerOffsetManager
    result = result && this.consumerOffsetManager.load();
    //加载subscriptionGroupManager
    result = result && this.subscriptionGroupManager.load();
    //加载messageStore
    if (result) {
        try {
            this.messageStore =
                new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                    this.brokerConfig);
            this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
            //load plugin
            MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
            this.messageStore = MessageStoreFactory.build(context, this.messageStore);
        } catch (IOException e) {
            result = false;
            e.printStackTrace();
        }
    }

    result = result && this.messageStore.load();
    ...
}

下面看一下this.topicConfigManager.load()的代码:

public boolean load() {
    String fileName = null;
    try {
        //读取文件路径
        fileName = this.configFilePath();
        //读取文件内容
        String jsonString = MixAll.file2String(fileName);
        if (null == jsonString || jsonString.length() == 0) {
            //读取备份文件
            return this.loadBak();
        } else {
            //将文件内容读入到topicConfigManager.topicConfigTable中
            this.decode(jsonString);
            PLOG.info("load {} OK", fileName);
            return true;
        }
    } catch (Exception e) {
        PLOG.error("load " + fileName + " Failed, and try to load backup file", e);
        return this.loadBak();
    }
}

consumerOffsetManager和subscriptionGroupManager的加载类似, 三个类都继承自ConfigManager.
接下去就是messageStore的加载, 用于消息的持久化, 关于这一部分, 可以参考前文.
3.当上述的管理器全部加载完成以后, 控制器将开始进入下一步的初始化：

• 启动netty服务端, 包括处理消息的remotingServer和主从同步使用的fastRemotingServer.
• 初始化多个线程池, 包括：sendMessageExecutor(处理发送消息), pullMessageExecutor(处理拉取消息), adminBrokerExecutor(管理Broker), clientManageExecutor(管理client), consumerManageExecutor(管理消费者)
• 将上述的线程池注册到netty消息处理器当中
• 启动定时调度线程来做很多事情, 包括：一天上报一次broker的所有状态, 10秒持久化一次Consumer消费进度, 60分钟清理一次无效的topic订阅信息, 60秒获取一次namesrv的地址信息

代码如下:

public boolean initialize() throws CloneNotSupportedException {
    ......
    if (result) {
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
        NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
        fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
        this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
        this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
            this.brokerConfig.getSendMessageThreadPoolNums(),
            this.brokerConfig.getSendMessageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.sendThreadPoolQueue,
            new ThreadFactoryImpl("SendMessageThread_"));

        this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
            this.brokerConfig.getPullMessageThreadPoolNums(),
            this.brokerConfig.getPullMessageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.pullThreadPoolQueue,
            new ThreadFactoryImpl("PullMessageThread_"));

        this.adminBrokerExecutor =
            Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                "AdminBrokerThread_"));

        this.clientManageExecutor = new ThreadPoolExecutor(
            this.brokerConfig.getClientManageThreadPoolNums(),
            this.brokerConfig.getClientManageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.clientManagerThreadPoolQueue,
            new ThreadFactoryImpl("ClientManageThread_"));

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

        this.registerProcessor();

        // TODO remove in future
        final long initialDelay = UtilAll.computNextMorningTimeMillis() - 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);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    BrokerController.this.consumerOffsetManager.persist();
                } catch (Throwable e) {
                    log.error("schedule persist consumerOffset error.", e);
                }
            }
        }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //清理出问题的consumer
                    BrokerController.this.protectBroker();
                } catch (Exception e) {
                    log.error("protectBroker error.", e);
                }
            }
        }, 3, 3, TimeUnit.MINUTES);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //打印生产, 消费队列信息
                    BrokerController.this.printWaterMark();
                } catch (Exception e) {
                    log.error("printWaterMark error.", e);
                }
            }
        }, 10, 1, TimeUnit.SECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
                } catch (Throwable e) {
                    log.error("schedule dispatchBehindBytes error.", e);
                }
            }
        }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);

        //更新namesrv的地址信息
        if (this.brokerConfig.getNamesrvAddr() != null) {
            this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
            log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
        } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
                    } catch (Throwable e) {
                        log.error("ScheduledTask fetchNameServerAddr exception", e);
                    }
                }
            }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
        }

        //slave更新master地址, 从master同步消息数据
        if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
            if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
                this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
                this.updateMasterHAServerAddrPeriodically = false;
            } else {
                this.updateMasterHAServerAddrPeriodically = true;
            }

            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.slaveSynchronize.syncAll();
                    } catch (Throwable e) {
                        log.error("ScheduledTask syncAll slave exception", e);
                    }
                }
            }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
        } else {
            //master打印slave落后的信息
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                        BrokerController.this.printMasterAndSlaveDiff();
                    } catch (Throwable e) {
                        log.error("schedule printMasterAndSlaveDiff error.", e);
                    }
                }
            }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
        }
    }
    return result;
}

上述的全部过程中, 比较关键的操作是: 启动netty服务端, 注册多个消息处理器, 初始化线程池用来做消息的并发处理.

start

当broker初始化了配置参数以后, 就可以开始启动部分了, 启动的部分会简单一些：

• 启动刚刚初始化的各个管理器：topicManager, consumerOffsetManager, subscriptionGroupManager, messageStore
• 开启定时调度线程30秒向namesrv不断上报自己的信息

public void start() throws Exception {
    if (this.messageStore != null) {
        this.messageStore.start();
    }

    if (this.remotingServer != null) {
        this.remotingServer.start();
    }

    if (this.fastRemotingServer != null) {
        this.fastRemotingServer.start();
    }

    if (this.brokerOuterAPI != null) {
        this.brokerOuterAPI.start();
    }

    if (this.pullRequestHoldService != null) {
        this.pullRequestHoldService.start();
    }

    if (this.clientHousekeepingService != null) {
        this.clientHousekeepingService.start();
    }

    if (this.filterServerManager != null) {
        this.filterServerManager.start();
    }

    this.registerBrokerAll(true, false);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                //向nameserver上报信息
                BrokerController.this.registerBrokerAll(true, false);
            } catch (Throwable e) {
                log.error("registerBrokerAll Exception", e);
            }
        }
    }, 1000 * 10, 1000 * 30, TimeUnit.MILLISECONDS);

    if (this.brokerStatsManager != null) {
        this.brokerStatsManager.start();
    }

    if (this.brokerFastFailure != null) {
        this.brokerFastFailure.start();
    }
}

后文介绍消息的接收.

参考资料
1.rocketMQ的broker模块