前面学习了name server的主要代码，这篇文章开始学习broker的源码。broker是RocketMQ的核心模块，这篇文章我们先从整体看下代码结构、RocketMQ的领域模型，然后再看下一个broker节点的启动过程以及Controller暴露的接口。
代码结构
broker的代码结构
领域模型
MQ领域语言描述RocketMQ做的事情，producer构建Message，发送给broker的指定topic，broker负责将消息投递到指定topic下的队列，并记录消息队列的offset，consumer利用拉模式拉取消息进行消费。
ddd-for-rmq.png
启动过程
BrokerController
BrokerController是broker模块的核心控制类，负责broker的初始化、启动、停止、资源管理，以及接受外部的请求并作出相应的动作。看下BrokerController中主要的属性，借此可以看下broker的基本功能
主要属性
- BrokerConfig，用于维护broker的配置信息
- NettyServerConfig，对于producer和consumer来说，broker是服务端
- NettyClientConfig，对于name server来说，broker是客户端
- MessageStoreConfig，消息存储的配置，RocketMQ一个非常厉害的特性就是上亿消息的堆积能力，堆积的消息是存储在broker的磁盘上的，那么这个类就是维护broker的消息存储的配置信息
- ConsumerManager，消费者管理
- ConsumerFilterManager，消费者消息过滤管理
- ProducerManager，生产者管理
- MessageArrivingListener，消息到达监听器
- BrokerOuterAPI，broker和外部系统沟通的适配层，有几个功能：（1）和name server交互，进行broker节点的注册和取消；（2）和其他broker节点交互；
上面这些不是全部，除此之外，还有几个线程池和线程池对应的队列，以及用于做HA的管理模块。显然，broker功能非常多，我们在接下来的几篇中慢慢梳理其中的代码。
public class BrokerController {
private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final Logger LOG_PROTECTION = LoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);
private static final Logger LOG_WATER_MARK = LoggerFactory.getLogger(LoggerName.WATER_MARK_LOGGER_NAME);
/**
* Broker的配置
*/
private final BrokerConfig brokerConfig;
/**
* netty服务端配置，对于生产者和消费者来说，broker是服务端
*/
private final NettyServerConfig nettyServerConfig;
/**
* netty客户端，对于name server来说，broker是客户端
*/
private final NettyClientConfig nettyClientConfig;
/**
* 消息存储配置
*/
private final MessageStoreConfig messageStoreConfig;
/**
* 消费者的offset管理
*/
private final ConsumerOffsetManager consumerOffsetManager;
/**
* 消费者管理
*/
private final ConsumerManager consumerManager;
/**
* 消费过滤管理
*/
private final ConsumerFilterManager consumerFilterManager;
/**
* 生产者管理
*/
private final ProducerManager producerManager;
/**
* 监听客户端和broker建立的通信通道，当通道关闭时候清理信息
*/
private final ClientHousekeepingService clientHousekeepingService;
/**
* 拉取消息处理器
*/
private final PullMessageProcessor pullMessageProcessor;
/**
* ???暂时不理解
*/
private final PullRequestHoldService pullRequestHoldService;
/**
* 消息到达监听器
*/
private final MessageArrivingListener messageArrivingListener;
/**
* 用于broker对client发起指令
*/
private final Broker2Client broker2Client;
private final SubscriptionGroupManager subscriptionGroupManager;
private final ConsumerIdsChangeListener consumerIdsChangeListener;
/**
* 负载均衡管理器
*/
private final RebalanceLockManager rebalanceLockManager = new RebalanceLockManager();
/**
* broker对外暴露的API
*/
private final BrokerOuterAPI brokerOuterAPI;
/**
* 调度线程池
*/
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
"BrokerControllerScheduledThread"));
/**
* 子节点同步器
*/
private final SlaveSynchronize slaveSynchronize;
/**
* 发送消息的线程池任务队列
*/
private final BlockingQueue<Runnable> sendThreadPoolQueue;
/**
* 拉取消息的线程池任务队列
*/
private final BlockingQueue<Runnable> pullThreadPoolQueue;
private final BlockingQueue<Runnable> queryThreadPoolQueue;
private final BlockingQueue<Runnable> clientManagerThreadPoolQueue;
private final BlockingQueue<Runnable> consumerManagerThreadPoolQueue;
private final FilterServerManager filterServerManager;
private final BrokerStatsManager brokerStatsManager;
private final List<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
private final List<ConsumeMessageHook> consumeMessageHookList = new ArrayList<ConsumeMessageHook>();
private MessageStore messageStore;
private RemotingServer remotingServer;
private RemotingServer fastRemotingServer;
private TopicConfigManager topicConfigManager;
private ExecutorService sendMessageExecutor;
private ExecutorService pullMessageExecutor;
private ExecutorService queryMessageExecutor;
private ExecutorService adminBrokerExecutor;
private ExecutorService clientManageExecutor;
private ExecutorService consumerManageExecutor;
private boolean updateMasterHAServerAddrPeriodically = false;
private BrokerStats brokerStats;
private InetSocketAddress storeHost;
private BrokerFastFailure brokerFastFailure;
private Configuration configuration;
//省略其他代码
}
initialize
这个方法用于初始化broker节点，主要的工作可以列举如下：
-
加载主要模块的配置信息
这个部分，会从外存总加载各个模块的配置信息，包括：topicConfigManager、consumerOffsetManager、subscriptionGroupManager、consumerFilterManager、messageStore。这里代码写得非常漂亮，使用了配置外化的思路和实现、应用了模板设计模式、插件设计模式和工厂设计模式。
- 模板设计模式
RocketMQ中的模板方法设计模式
-
插件设计模式
插件设计模式和工厂设计模式一起使用，需要包含一个插件上下文、一个抽象插件类（AbstractPluginMessageStore），主要模块入下图所示。
插件设计模式
插件上下文（MessageStorePluginContext）用于保存跟插件相关的信息，看下插件上下文的代码：
public class MessageStorePluginContext { private MessageStoreConfig messageStoreConfig; private BrokerStatsManager brokerStatsManager; private MessageArrivingListener messageArrivingListener; private BrokerConfig brokerConfig; public MessageStorePluginContext(MessageStoreConfig messageStoreConfig, BrokerStatsManager brokerStatsManager, MessageArrivingListener messageArrivingListener, BrokerConfig brokerConfig) { super(); this.messageStoreConfig = messageStoreConfig; this.brokerStatsManager = brokerStatsManager; this.messageArrivingListener = messageArrivingListener; this.brokerConfig = brokerConfig; } //省略了getter和setter方法 } -
工厂设计模式
public final class MessageStoreFactory { public final static MessageStore build(MessageStorePluginContext context, MessageStore messageStore) throws IOException { //从配置文件中取出配置好的插件 String plugin = context.getBrokerConfig().getMessageStorePlugIn(); if (plugin != null && plugin.trim().length() != 0) { String[] pluginClasses = plugin.split(","); //依次加载插件类对象，并生成对应的MessageStore对象 for (int i = pluginClasses.length - 1; i >= 0; --i) { String pluginClass = pluginClasses[i]; try { @SuppressWarnings("unchecked") Class<AbstractPluginMessageStore> clazz = (Class<AbstractPluginMessageStore>) Class.forName(pluginClass); Constructor<AbstractPluginMessageStore> construct = clazz.getConstructor(MessageStorePluginContext.class, MessageStore.class); messageStore = construct.newInstance(context, messageStore); } catch (Throwable e) { throw new RuntimeException(String.format( "Initialize plugin's class %s not found!", pluginClass), e); } } } return messageStore; } }
-
启动服务器
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.queryMessageExecutor = new BrokerFixedThreadPoolExecutor( this.brokerConfig.getQueryMessageThreadPoolNums(), this.brokerConfig.getQueryMessageThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS, this.queryThreadPoolQueue, new ThreadFactoryImpl("QueryMessageThread_")); 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_")); -
注册请求处理器
这个方法类似于name server里的用法，这里不仔细展开讲
-
设置各种定时任务，包括：获取broker状态的、周期性将消费者的offset刷到硬盘、周期性检查消费者的消费能力以保护broker、定期打印消息消费标记等等。
-
获取name server的地址
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节点的不同处理
如果当前节点是slave节点，则设置一个定时任务：每隔一段时间，就将配置信息从master节点同步到当前节点；如果当前节点是master节点，则设置一个定时任务：每隔一段时间，就对比master节点和slave节点的配置信息，并打印出不相同的配置。
其他
其他还有start、shutdown和registerBrokerAll等方法，其中reigsterBrokerAll方法的作用是将broker节点注册到name server，这样producer和consumer就可以拿到broker节点的地址信息。
