附录G Netty与NettyUtils

注：本文是为了配合《Spark内核设计的艺术 架构设计与实现》一书的内容而编写，目的是为了节省成本、方便读者查阅。书中附录G的内容都在本文呈现。

Netty是一个NIO客户端服务器框架，使得开发高性能、高可靠性的网络服务器和客户端程序变得快速且容易。它极大地简化

化了网络编程，如TCP和UDP套接字服务器。

“快速和容易”并不意味着应用程序会有难维护和性能低的问题。Netty是一个精心设计的框架，它从许多协议的实现中吸收了很多的经验，比如FTP, SMTP, HTTP和多种多样的二进制和基于文本的传统协议。因此Netty在不降低开发效率、性能、稳定性、灵活性情况下，已经成功地找到了解决方法。

NettyUtils是Spark对Netty API的封装，在这里对其中的主要方法进行说明。有关Netty的更多内容请访问Netty官网http://netty.io/。

createThreadFactory

功能描述：创建线程工厂，此工厂生成的线程都使用给定的前缀名threadPoolPrefix + "-" +数字的格式命名。

  public static ThreadFactory createThreadFactory(String threadPoolPrefix) {
    return new DefaultThreadFactory(threadPoolPrefix, true);
  }

createEventLoop

功能描述：根据参数IOMode，创建Netty的EventLoopGroup。

  public static EventLoopGroup createEventLoop(IOMode mode, int numThreads, String threadPrefix) {
    ThreadFactory threadFactory = createThreadFactory(threadPrefix);

    switch (mode) {
      case NIO:
        return new NioEventLoopGroup(numThreads, threadFactory);
      case EPOLL:
        return new EpollEventLoopGroup(numThreads, threadFactory);
      default:
        throw new IllegalArgumentException("Unknown io mode: " + mode);
    }
  }

getClientChannelClass

功能描述：根据参数IOMode，返回正确的客户端SocketChannel。

  public static Class<? extends Channel> getClientChannelClass(IOMode mode) {
    switch (mode) {
      case NIO:
        return NioSocketChannel.class;
      case EPOLL:
        return EpollSocketChannel.class;
      default:
        throw new IllegalArgumentException("Unknown io mode: " + mode);
    }
  }

getServerChannelClass

功能描述：根据参数IOMode，返回正确的服务端SocketChannel。

  public static Class<? extends ServerChannel> getServerChannelClass(IOMode mode) {
    switch (mode) {
      case NIO:
        return NioServerSocketChannel.class;
      case EPOLL:
        return EpollServerSocketChannel.class;
      default:
        throw new IllegalArgumentException("Unknown io mode: " + mode);
    }
  }

createFrameDecoder

功能描述：创建一个LengthFieldBasedFrameDecoder。LengthFieldBasedFrameDecoder的5个参数分别代表frame的最大长度、长度字段的偏移量、长度字段的字节数、需要排除的长度字段的字节数、长度字段的初始长度。所以创建的LengthFieldBasedFrameDecoder的前8个字节代表frame的长度。LengthFieldBasedFrameDecoder通常会被设置到SocketChannel的管道中，在所有Decoder被调用之前调用。

  public static TransportFrameDecoder createFrameDecoder() {
    return new TransportFrameDecoder();
  }

getRemoteAddress

功能描述：返回Channel的远端地址。

  public static String getRemoteAddress(Channel channel) {
    if (channel != null && channel.remoteAddress() != null) {
      return channel.remoteAddress().toString();
    }
    return "<unknown remote>";
  }

createPooledByteBufAllocator

功能描述：创建一个汇集ByteBuf但对本地线程缓存禁用的分配器。为什么要对本地线程缓存禁用？因为ByteBuf都是由事件循环线程分配，所以线程本地缓存对于TransportClient是禁用的。但是ByteBuf的释放却是由Executor线程，而不是事件循环线程来完成。本地线程缓存经常会延迟ByteBuf的回收，导致巨大的内存消耗。

  public static PooledByteBufAllocator createPooledByteBufAllocator(
      boolean allowDirectBufs,
      boolean allowCache,
      int numCores) {
    if (numCores == 0) {
      numCores = Runtime.getRuntime().availableProcessors();
    }
    return new PooledByteBufAllocator(
      allowDirectBufs && PlatformDependent.directBufferPreferred(),
      Math.min(getPrivateStaticField("DEFAULT_NUM_HEAP_ARENA"), numCores),
      Math.min(getPrivateStaticField("DEFAULT_NUM_DIRECT_ARENA"), allowDirectBufs ? numCores : 0),
      getPrivateStaticField("DEFAULT_PAGE_SIZE"),
      getPrivateStaticField("DEFAULT_MAX_ORDER"),
      allowCache ? getPrivateStaticField("DEFAULT_TINY_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_SMALL_CACHE_SIZE") : 0,
      allowCache ? getPrivateStaticField("DEFAULT_NORMAL_CACHE_SIZE") : 0
    );
  }

getPrivateStaticField

功能描述：用于获得Netty的静态属性值 。

  private static int getPrivateStaticField(String name) {
    try {
      Field f = PooledByteBufAllocator.DEFAULT.getClass().getDeclaredField(name);
      f.setAccessible(true);
      return f.getInt(null);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }