enoc

概述

在上一节 RocketMQ源码详解 | Producer篇 · 其二:消息组成、发送链路 中,我们终于将消息发送出了 Producer,在短暂的 tcp 握手后,很快它就会进入目的 Broker。这次我们来自底向上的看下 Broker 端是如何接收然后分发处理消息,同时了解 RocketMQ 的 Broker 的线程模型。


Netty 组件

如果你还记得上一节的内容的话那应该知道,NettyRomotingAbstract 有两个实现类,分别是 NettyRemotingClientNettyRemotingServer ,我们已经知道了前者的实现,现在我们再来看看后者


NettyRemotingServer

这个类很长,我们先来看它的属性

/*    引导类和dispatch线程与select线程池   */
private final ServerBootstrap serverBootstrap;
private final EventLoopGroup eventLoopGroupSelector;
private final EventLoopGroup eventLoopGroupBoss;
// 配置类
private final NettyServerConfig nettyServerConfig;

// 用来执行 callback 函数的线程池
private final ExecutorService publicExecutor;
// 自定义的 Channel 事件监听器
private final ChannelEventListener channelEventListener;

// 扫描已经超时的 ResponseFeature
private final Timer timer = new Timer("ServerHouseKeepingService", true);
// 工作线程
private DefaultEventExecutorGroup defaultEventExecutorGroup;


private int port = 0;

private static final String HANDSHAKE_HANDLER_NAME = "handshakeHandler";
private static final String TLS_HANDLER_NAME = "sslHandler";
private static final String FILE_REGION_ENCODER_NAME = "fileRegionEncoder";

// sharable handlers
private HandshakeHandler handshakeHandler;
private NettyEncoder encoder;
private NettyConnectManageHandler connectionManageHandler;
private NettyServerHandler serverHandler;

我们主要关心 serverBootStrap 的启动


首先是它的初始化,初始化代码较长,主要做了三件事:

  1. 初始化 callback 函数执行线程池
  2. 在 Linux 平台上启用 epoll
  3. 使用可能存在的 SSL

然后是重头戏,其具体的创建

ServerBootstrap childHandler =
  this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
  .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
  // 半连接队列长度
  .option(ChannelOption.SO_BACKLOG, 1024)
  // 开启内核中的 net.ipv4.tcp_tw_reuse 选项
  .option(ChannelOption.SO_REUSEADDR, true)
  // 关闭操作系统的连接维护,由自己去干
  .option(ChannelOption.SO_KEEPALIVE, false)
  // 禁用 Nagle 算法
  .childOption(ChannelOption.TCP_NODELAY, true)
  // 设定发送缓冲区和接收缓冲区大小
  .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
  .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
  // 设置监听端口(0.0.0.0:xx)
  .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
  .childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
      ch.pipeline()
        // 设置握手处理器
        .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
        .addLast(defaultEventExecutorGroup,
                 // 设置编解码器
                 encoder,
                 new NettyDecoder(),
                 // 注册 Netty 的心跳检查
                 new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                 // 管理连接,超时处理,维护channelTables与存活的连接
                 connectionManageHandler,
                 // 实际上的处理收到的请求
                 serverHandler
                );
    }
  });

这里需要关注的点很多,我们按照顺序来看

首先是线程模型,在这里我们可以看出它是 1(eventLoopGroupBoss) - N(eventLoopGroupSelector) - M(defaultEventExecutorGroup) 的线程模型,即有 一个 Acceptor,N 个 Select 线程,和 M 个 IO 线程。

如果了解过 Reactor 模型的话可以看出这属于主从多 Reactor 模式,在 Nginx、Kakfa、Tomcat 都能看到类似的设计。


然后需要关注的是 SO_BACKLOG,这里指定了半队列的长度为 1024

backlog

在 TCP 的三次握手中,backlog 用于处理从 SYN RECEIVED 到 ESTABLISHED 状态之间的套接字。

其中具有 SYN 队列和 accept 队列:

  • SYN 队列

    长度由系统调整。

    当服务器端收到一个 SYN 包时,将其放入 SYN 队列并返回 ACK+SYN。队满则抛弃,客户端超时后重发。

  • accept 队列

    长度由程序调整(也就是我们通过 SO_BACKLOG 设置的长度)。

    当服务器端收到之前自己发送的 SYN 的 ACK 时,会将套接字放入这里。大多数时候这里的数据可以很快的被程序通过 accept() 取出。队满时抛弃到来的 ACK 包(虽然客户端已经进入了 ESTABLISHED 状态,但由于 tcp 的慢启动,并不会造成太大影响),客户端重发到一定次数仍未被放入 accept 队列时会被发送 RST 包。同时在 Linux 中,这里队满时会对 SYN 队列的接收速率进行控制。


再通过 SO_REUSEADDR 开启了内核的 net.ipv4.tcp_tw_reuse 选项

net.ipv4.tcp_tw_reuse

这个选项主要用在具有大量短连接的应用。

问题:

在具有大量短连接时,服务器端上具有太多属于同一个客户端的处于 TIME_WAIT 状态的连接,而导致该客户端不能建立新的连接。

处理方法:

在 Linux 中,TCP 的 TIME_WAIT 时间默认为 1 分钟,而 TIME_WAIT 被设计出来的主要目的有两个:

  1. 避免新的连接收到旧的连接的重发数据包
  2. 确保远程端不是在 LAST_ACK 状态

在开启这个选项后,如果 TIME_WAIT 状态的连接过多,会使用在 TCP 可选头部中的时间戳选项,来和之前存储的时间戳对比,若该大,则从 TIME_WAIT 状态的存活连接中随机选取一个并分配给该 TCP 连接。

对于需要解决问题 1,由于旧的连接的重发包具有过期的时间戳,所以会被丢弃;

对于问题 2 ,当处于 LAST_ACK 的一端收到新的 TCP 连接的 SYN 包后,会将其丢弃,然后重发 FIN 包,处于 SYN_SEND 状态的一端收到这种错误的包后会发送 RST 包,然后再发送 SYN 包重试。


然后使用 SO_KEEPALIVE 关闭操作系统自带的 KeepAlive 机制。

这是因为操作系统的连接维护默认为 2 小时,对其修改需要系统调用,且当协议被切换为 UDP 时会失效,故我们在后面使用了 IdleStateHandler 来注册 Netty 自己实现的心跳检测


接着将 TCP_NODELAY 设置为 True 来禁用 Nagle 算法。

这是因为 Nagle 算法会等待当前 TCP 的包到达了足够的大小才会发送,这会造成发送延迟


再往后看可以发现是先注册了 HandshakeHandler,我们来看它干了什么

@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  // 标记当前位置以便恢复。因为我们接下来需要查看第一个字节以确定内容是否以 TLS 握手开始
  msg.markReaderIndex();

  byte b = msg.getByte(0);

  // 握手的魔数,如果是说明这是个tls握手
  if (b == HANDSHAKE_MAGIC_CODE) {
    switch (tlsMode) {
        // 禁用 SSL
      case DISABLED:
        ctx.close();
        log.warn("Clients intend to establish an SSL connection while this server is running in SSL disabled mode");
        break;
        // 可用或必须使用 SSL
      case PERMISSIVE:
      case ENFORCING:
        if (null != sslContext) {
          // 添加 SSL handler
          ctx.pipeline()
            // SSL 隧道
            .addAfter(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, TLS_HANDLER_NAME, sslContext.newHandler(ctx.channel().alloc()))
            // 用来保证文件在零拷贝时也进入能被 SSL 加密
            .addAfter(defaultEventExecutorGroup, TLS_HANDLER_NAME, FILE_REGION_ENCODER_NAME, new FileRegionEncoder());
          log.info("Handlers prepended to channel pipeline to establish SSL connection");
        } else {
          ctx.close();
          log.error("Trying to establish an SSL connection but sslContext is null");
        }
        break;

      default:
        log.warn("Unknown TLS mode");
        break;
    }
  } else if (tlsMode == TlsMode.ENFORCING) {
    ctx.close();
    log.warn("Clients intend to establish an insecure connection while this server is running in SSL enforcing mode");
  }

  // 恢复read索引,以便握手协商可以正常进行。
  msg.resetReaderIndex();

  try {
    // 完成 SSL 的判定后将被于本 pipeline 中移除
    ctx.pipeline().remove(this);
  } catch (NoSuchElementException e) {
    log.error("Error while removing HandshakeHandler", e);
  }

  // 交给下一个 handler
  ctx.fireChannelRead(msg.retain());
}

从代码我们可以知道,这个 Handler 用于判断是否使用 SSL 对连接进行加密,有的话则使用


然后是我们之前提到过的 IdleStateHandler ,它的几个参数分别是:

  1. 读超时时间
  2. 写超时时间
  3. 读写超时时间

而我们在这将 1 和 2 都设置为了 0,即不进行触发

一旦超时,它将会产生 IdleStateEvent ,在下一个 Handler NettyConnectManageHandler 中,我们可以看到它被捕获了

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  if (evt instanceof IdleStateEvent) {
    IdleStateEvent event = (IdleStateEvent) evt;
    if (event.state().equals(IdleState.ALL_IDLE)) {
      final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
      log.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
      RemotingUtil.closeChannel(ctx.channel());
      if (NettyRemotingServer.this.channelEventListener != null) {
        NettyRemotingServer.this
          .putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
      }
    }
  }

  ctx.fireUserEventTriggered(evt);
}

最后其他的组件都和上一章讲过差不多,故不再重复。接下来主要看一个和 Client 不同的地方。


ChannelEventListener

在上一章了解 Client 时,NettyConnectManageHandler 中在每一个状态中都有以下代码

if (NettyRemotingServer.this.channelEventListener != null) {
  NettyRemotingServer.this
    .putNettyEvent(new NettyEvent(NettyEventType./*  XXX  */, remoteAddress, ctx.channel()));
}

Client 由于没有注册 channelEventListener 而没有使用,在 NettyRemotingServer 中则在执行构造器时注册了 ClientHousekeepingService ,当然是 Broekr 端,还有一个是 BrokerHousekeepingService ,用于 NameServer

public void start() {
  this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      try {
        ClientHousekeepingService.this.scanExceptionChannel();
      } catch (Throwable e) {
        log.error("Error occurred when scan not active client channels.", e);
      }
    }
  }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
}

private void scanExceptionChannel() {
  this.brokerController.getProducerManager().scanNotActiveChannel();
  this.brokerController.getConsumerManager().scanNotActiveChannel();
  this.brokerController.getFilterServerManager().scanNotActiveChannel();
}

从实现就能看出来,这个类是在定期扫描过期的 Channel 并移除,同时通过监听事件在其 close、exception、idle 时移除


NettyRemotingAbstract

最后回到 NettyRemotingAbstractprocessRequestCommand 方法,虽然在上一节中已经看过了,不过我们再来详细看一次

final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;

首先我们可以知道在 processorTable 中存放着响应码和其对应的请求处理器与执行线程池,如果没有会使用默认处理器。

然后是使用其对应的线程池来执行业务请求,并使用处理回调函数

try {
  doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
  final RemotingResponseCallback callback = response -> { /* xxx */ };
  // 如果是异步请求处理器,则将回调函数交给其
  if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
    AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
    processor.asyncProcessRequest(ctx, cmd, callback);
  } else {
    NettyRequestProcessor processor = pair.getObject1();
    RemotingCommand response = processor.processRequest(ctx, cmd);
    // 否则进行同步的调用
    callback.callback(response);
  }
} catch (Throwable e) {
	/* xxx  */
}

那么,这些响应函数和线程池是在什么时候放入的呢?通过追踪,我们发现了 BrokerController 类,其在初始化时调用的 registerProcessor 函数如下:

// 用于处理消息的发送请求
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
 */
this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);

/**
 * ReplyMessageProcessor
 */
ReplyMessageProcessor replyMessageProcessor = new ReplyMessageProcessor(this);
replyMessageProcessor.registerSendMessageHook(sendMessageHookList);

this.remotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE, replyMessageProcessor, replyMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE_V2, replyMessageProcessor, replyMessageExecutor);
this.fastRemotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE, replyMessageProcessor, replyMessageExecutor);
this.fastRemotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE_V2, replyMessageProcessor, replyMessageExecutor);

/*  以下略   */

我们主要观察到了几个重点:

  1. 每一类业务处理都由该业务类型对应的线程池来处理

  2. 同时维护 remotingServer 和 fastRemotingServer 两个处理服务

    如果你对在第一节提到过的 VIP 还有印象的话,应该可以想起 VIP 端口就是 普通端口号-2。而这里的 fastRemotingServer,监控的就是 VIP 端口


至此,我们终于可以画出 RocketMQ 在 Broker 端的线程模型了


相关文章:

猜你喜欢