public interface RemotingService {
  	// 启动 Netty Boss、Worker 事件循环组
    void start();
		// 释放对应的线程资源，优雅退出等
    void shutdown();
		// 注册类似拦截器的功能，在请求前、后执行对应的操作
    void registerRPCHook(RPCHook rpcHook);
}

public interface RemotingServer extends RemotingService {
    // 注册请求编码器对应处理器，并交给对应的 ExecutorService 执行
    void registerProcessor(final int requestCode, final NettyRequestProcessor processor,
        final ExecutorService executor);

    // 注册默认请求编码器对应处理器，并交给对应的 ExecutorService 执行
    void registerDefaultProcessor(final NettyRequestProcessor processor, final ExecutorService executor);

    // 获取请求编码器对应处理器，并交给对应的 ExecutorService 执行
    Pair<NettyRequestProcessor, ExecutorService> getProcessorPair(final int requestCode);

    // 执行同步请求
    RemotingCommand invokeSync(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) throws InterruptedException, RemotingSendRequestException,
        RemotingTimeoutException;

    // 执行异步请求
    void invokeAsync(final Channel channel, final RemotingCommand request, final long timeoutMillis,
        final InvokeCallback invokeCallback) throws InterruptedException,
        RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException;

    // 执行单向请求
    void invokeOneway(final Channel channel, final RemotingCommand request, final long timeoutMillis)
        throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException,
        RemotingSendRequestException;

}

public interface RemotingClient extends RemotingService {
    // 更新 NameServer 地址
    void updateNameServerAddressList(final List<String> addrs);

    // 获取 NameServer 地址
    List<String> getNameServerAddressList();

    // 执行同步请求
    RemotingCommand invokeSync(final String addr, final RemotingCommand request,
        final long timeoutMillis) throws InterruptedException, RemotingConnectException,
        RemotingSendRequestException, RemotingTimeoutException;

    // 执行异步请求
    void invokeAsync(final String addr, final RemotingCommand request, final long timeoutMillis,
        final InvokeCallback invokeCallback) throws InterruptedException, RemotingConnectException,
        RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException;

    // 执行单向请求
    void invokeOneway(final String addr, final RemotingCommand request, final long timeoutMillis)
        throws InterruptedException, RemotingConnectException, RemotingTooMuchRequestException,
        RemotingTimeoutException, RemotingSendRequestException;

    // 注册处理器
    void registerProcessor(final int requestCode, final NettyRequestProcessor processor,
        final ExecutorService executor);
}

// 如下方法也就是官网中介绍的 Header 字段
// remark、ext 字段需要额外计算
private static int calTotalLen(int remark, int ext) {
  // int code(~32767)
  int length = 2
    // LanguageCode language
    + 1
    // int version(~32767)
    + 2
    // int opaque
    + 4
    // int flag
    + 4
    // String remark
    + 4 + remark
    // HashMap<String, String> extFields
    + 4 + ext;
  return length;
}

private boolean useEpoll() {
  // OS 类型：Windows、Linux
  return RemotingUtil.isLinuxPlatform()
    // 通过 NettyServerConfig.setUseEpollNativeSelector 方法设置是否开启 Epoll Selector 模型
    && nettyServerConfig.isUseEpollNativeSelector()
    && Epoll.isAvailable();
}

// public NettyRemotingServer(final NettyServerConfig nettyServerConfig, final ChannelEventListener channelEventListener) {
if (useEpoll()) {
  this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() {
    private AtomicInteger threadIndex = new AtomicInteger(0);

    @Override
    public Thread newThread(Runnable r) {
      return new Thread(r, String.format("NettyEPOLLBoss_%d", this.threadIndex.incrementAndGet()));
    }
  });

  this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
    private AtomicInteger threadIndex = new AtomicInteger(0);
    private int threadTotal = nettyServerConfig.getServerSelectorThreads();

    @Override
    public Thread newThread(Runnable r) {
      return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
    }
  });
} else {
  this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
    private AtomicInteger threadIndex = new AtomicInteger(0);

    @Override
    public Thread newThread(Runnable r) {
      return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
    }
  });

  this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
    private AtomicInteger threadIndex = new AtomicInteger(0);
    private int threadTotal = nettyServerConfig.getServerSelectorThreads();

    @Override
    public Thread newThread(Runnable r) {
      return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
    }
  });
}

/*
 * SslHandler：SSL安全套接字协议
 * ⬇
 * FileRegionEncoder：文件区域采用 Zero-Copy SendFile 编码传输
 * ⬇
 * NettyEncoder：编码器
 * ⬇
 * NettyDecoder：解码器
 * ⬇
 * IdleStateHandler：空闲检查
 * ⬇
 * NettyConnectManageHandler：网络连接管理
 * ⬇
 * NettyServerHandler：服务端请求处理器
 */
ServerBootstrap childHandler =
  this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
  .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
  .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getServerSocketBacklog())
  .option(ChannelOption.SO_REUSEADDR, true)
  .option(ChannelOption.SO_KEEPALIVE, false)
  .childOption(ChannelOption.TCP_NODELAY, true)
  .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(),
                 new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                 connectionManageHandler,
                 serverHandler
                );
    }
  });

// Math.min(Runtime.getRuntime().availableProcessors(), 4)
this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
  this.brokerConfig.getSendMessageThreadPoolNums(),
  this.brokerConfig.getSendMessageThreadPoolNums(),
  1000 * 60,
  TimeUnit.MILLISECONDS,
  this.sendThreadPoolQueue,
  new ThreadFactoryImpl("SendMessageThread_"));

/*
 * SendMessageProcessor、sendMessageExecutor
 */
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);