6.1 多端口监听

2019-12-28 14:00:05 14,920 4

本文主要介绍RocketMQ的多端口监听机制,通过本文,你可以了解到Broker端源码中remotingServer和fastRemotingServer的区别,以及客户端配置中,vipChannelEnabled的作用。

1 多端口监听

在RocketMQ中,可以通过broker.conf配置文件中指定listenPort配置项来指定Broker监听客户端请求的端口,如果不指定,默认监听10911端口。

listenPort=10911

不过,Broker启动时,实际上会监听3个端口:10909、10911、10912,如下所示:

$ lsof -iTCP -nP | grep LISTEN
java  1892656 tianshouzhi.robin   96u  IPv6 14889281  0t0  TCP *:10912 (LISTEN)
java  1892656 tianshouzhi.robin  101u  IPv6 14889285  0t0  TCP *:10911 (LISTEN)
java  1892656 tianshouzhi.robin  102u  IPv6 14889288  0t0  TCP *:10909 (LISTEN)

而其他两个端口是根据listenPort的值,动态计算出来的。这三个端口由Broker内部不同的组件使用,作用分别如下:

    remotingServer:监听listenPort配置项指定的监听端口,默认10911

    fastRemotingServer:监听端口值listenPort-2,即默认为10909

    HAService:监听端口为值为listenPort+1,即10912,该端口用于Broker的主从同步


本文主要聚焦于remotingServer和fastRemotingServer的区别:

    Broker端:remotingServer可以处理客户端所有请求,如:生产者发送消息的请求,消费者拉取消息的请求。fastRemotingServer功能基本与remotingServer相同,唯一不同的是不可以处理消费者拉取消息的请求。Broker在向NameServer注册时,只会上报remotingServer监听的listenPort端口。

    客户端:默认情况下,生产者发送消息是请求fastRemotingServer,我们也可以通过配置让其请求remotingServer;消费者拉取消息只能请求remotingServer。

下面通过源码进行验证Broker端构建remotingServer和fastRemotingServer时的区别,以及客户端如何配置。

2 Broker端

BrokerController内部定义了remotingServer和fastRemotingServer两个字段

private RemotingServer remotingServer;
private RemotingServer fastRemotingServer;

在初始化时,在initiallize方法内部会对这两个字段的进行初始化:

BrokerController#initialize

public boolean initialize() throws CloneNotSupportedException {
    boolean result = this.topicConfigManager.load();

    result = result && this.consumerOffsetManager.load();
    result = result && this.subscriptionGroupManager.load();
    result = result && this.consumerFilterManager.load();

    if (result) {..}//加载message store,略

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

    if (result) {

        //1 remotingServer监听listenPort端口,默认10911
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, 
                                                      this.clientHousekeepingService);
                                                      
        //2 fastRemotingServer监听listenPort-2端口,默认10990
        NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
        fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
        this.fastRemotingServer = new NettyRemotingServer(fastConfig, 
                                                    this.clientHousekeepingService);

        //...启动异步线程池,略

        //3 注册请求处理器
        this.registerProcessor();

可以看到,这两个字段实例化时:remotingServer使用了nettyServerConfig配置;而fastRemotingServer将配置克隆了一份,然后只是修改了监听的的端口号,其他不变。

创建完之后remotingServer和fastRemotingServer,会调用registerProcessor注册请求处理器。fastRemotingServer与remotingServer注册的请求处理器类型几乎完全相同,相关源码如下红色框所示:

org.apache.rocketmq.broker.BrokerController#registerProcessor

public void registerProcessor() {
    /**
     * SendMessageProcessor
     */
    SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
    sendProcessor.registerSendMessageHook(sendMessageHookList);
    sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

    this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
    this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
    this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor,this.sendMessageExecutor);
    this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);

    this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
    this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);

    /**
     * PullMessageProcessor,注意这里只注册到了到了remotingServer中,没有注册到fastRemotingServer
     */
    this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
    this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
    
    //...


可以看到,唯一不同的是,对于PullMessageProcessor,只在remotingServer中注册了,并没有在fastRemotingServer注册。意味着为fastRemotingServer不可以处理消费者拉取消息的请求(还有很多其他的处理器类型,这里并没有完全列出)。

在明白了fastRemotingServer和remotingServer之后,下面从客户端分析,如何进行选择。

3 客户端

客户端的DefaultMQProducerDefaultMQPushConsumer都继承了ClientConfig类,这个类中有一些公共的配置项,其中包含一个布尔字段vipChannelEnabled。从字面意思看,其用于控制是否开启VIP通道,如果为true,生产者发送的消息会请求fastRemotingServer,否则请求remotingServer。

在RocketMQ 4.5.0及之前,vipChannelEnabled字段值默认为true。在RocketMQ 4.5.1之后,修改为了false。可以通过JVM参数 -Dcom.rocketmq.sendMessageWithVIPChannel=false,来修改这个默认值。

org.apache.rocketmq.client.ClientConfig

public class ClientConfig {
//是否启用VIP通道
  private boolean vipChannelEnabled = Boolean.parseBoolean(
                System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, 
                "true"));
...
  public boolean isVipChannelEnabled() {
      return vipChannelEnabled;
  }
  public void setVipChannelEnabled(final boolean vipChannelEnabled) {
      this.vipChannelEnabled = vipChannelEnabled;
  }
...
}


生产者:

生产者在发送消息时,都会通过DefaultMQProducerImpl#sendKernelImpl方法,这个方法内部会判断是否开启VIP通道,如下图红色框:

private SendResult sendKernelImpl(final Message msg,
    final MessageQueue mq,
    final CommunicationMode communicationMode,
    final SendCallback sendCallback,
    final TopicPublishInfo topicPublishInfo,
    final long timeout) throws MQClientException {
    long beginStartTime = System.currentTimeMillis();
    String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    if (null == brokerAddr) {
        tryToFindTopicPublishInfo(mq.getTopic());
        brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    }

    SendMessageContext context = null;
    //判断是否开启了VIP通道
    if (brokerAddr != null) {
        brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), 
                                             brokerAddr);
    //...

在开启VIP通道的情况下,会将请求的broker 端口地址-2,改为请求fastRemotingServer,如下所示:

org.apache.rocketmq.common.MixAll#brokerVIPChannel

public static String brokerVIPChannel(final boolean isChange, final String brokerAddr) {
    if (isChange) {
        String[] ipAndPort = brokerAddr.split(":");
        String brokerAddrNew = ipAndPort[0] + ":" 
                               + (Integer.parseInt(ipAndPort[1]) - 2);
        return brokerAddrNew;
    } else {
        return brokerAddr;
    }
}


消费者

消费者拉取消息总是会调用remotingServer,因为PullMessageProcessor只在remotingServer中进行了注册,fastRemotingServer无法处理这个请求,因此并不会修改端口,可参考PullAPIWrapper类。


关于其他请求:

Broker支持很多客户端请求类型,除了发送/拉取消息之外,还包括创建Topic、查询/更新offset,发送心跳信息,查询消费者组ID列表等。具体请求哪个端口,主要是看有没有调用MixAl#brokerVIPChannel方法修改端口。例如对于心跳请求,即使设置了brokerVIPChannel也不起作用,因为心跳请求之前不会修改端口号,因此总是请求remotingServer。