Netty4简单认知

Posted 2021-04-30 程序员TD的生活

Channel简介

在Netty中，Channel相当于一个Socket的抽象，它为用户提供了关于Socket状态（是连接还是断开）及对Socket的读、写等操作。每当Netty建立了一个连接，都创建一个与其对应的Channel实例。

Channel的注册过程所做的工作就是将Channel与对应的EventLoop进行关联。因此，在Netty中，每个Channel都会关联一个特定的EventLoop，并且这个Channel中的所有I/O操作都是在这个EventLoop中执行的；当关联好Channel和EventLoop后，会继续调用底层JavaNIO的SocketChannel对象的register()方法，将底层Java NIO的SocketChannel注册到指定的Selector中。通过这两步，就完成了Netty对Channel的注册过程。

而对于Channel的创建过程中，会传入参数Channel创建ChannelPipeline，创建ChannelPipeline时候会创建ChannelHandlerContext传入ChannelPipeline。

下图表示常用Channel：

NiosocketChannel的创建

    Bootstrap是Netty提供的一个便利的工厂类，可以通过它来完成客户端或服务端的Netty初始化。先来看一个例子，从客户端程序是如何启动的。首先，从客户端的代码片段开始。

package com.example.gateway;
import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;
public class NettyChatClient { public NettyChatClient connect(int port, String host, final String nickName) { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception {
 } }); ChannelFuture channelFuture = bootstrap.connect(host, port).sync(); channelFuture.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } return this; }}

在实例化NioSocketChannel的过程中，Unsafe就特别关键。Unsafe其实是对Java底层Socket操作的封装，因此，它实际上是沟通Netty上层和Java底层的重要桥梁。

• NioSocketChannel创建过程（只是将构造函数赋值给了bootstrapt，实例化过程需要在bootstrap.connect链接时构建）：

  (1)调用NioSocketChannel.newSocket（DEFAULT_SELECTOR_PROVIDER）打开一个新的Java NioSocketChannel。

（2）初始化AbstractChannel（Channel parent）对象并给属性赋值，具体赋值的属性如下。

      id：每个Channel都会被分配一个唯一的id。

      parent：属性值默认为null。

      unsafe：通过调用newUnsafe()方法实例化一个Unsafe对象，它的类型是AbstractNioByteChannel.NioByteUnsafe内部类。

      pipeline：是通过调用new DefaultChannelPipeline(this)新创建的实例，而unsafe。DefaultChannelPipeline中还有两个特殊的属性，即Head和Tail，这两个属性是双向链表的头和尾。其实在DefaultChannelPipeline中维护了一个以AbstractChannelHandlerContext为节点元素的双向链表，这个链表是Netty实现Pipeline机制的关键。

// DefaultChannelPipeline的依赖final AbstractChannelHandlerContext head;final AbstractChannelHandlerContext tail;// 构造函数的初始化tail = new TailContext(this);head = new HeadContext(this);

pipeline在connect的时候初始化init channel过程中注册被封装了ChannerHandler的AbstractChannelHandlerContext类，这个注册的过程是：tail.prev = newCtx;，之后在调用connect的时候会调用AbstractChannelHandlerContext的connect方法，最终调用的是Unsafe的connect方法

（3）AbstractNIOChannel中被赋值的属性如下。

      ch：被赋值为Java原生SocketChannel，即NioSocketChannel的newSocket()方法返回的Java NIO SocketChannel。

      readInterestOp：被赋值为SelectionKey.OP_READ。

      ch：被配置为非阻塞，即调用ch.configureBlocking(false)方法。

(4)NioSocketChannel中被赋值的属性：config=new NioSocketChannelConfig(this，socket.socket())。

• EventLoop的初始化

MultithreadEventLoopGroup作为NioEventLoopGroup父类，基本操作都在MultithreadEventLoopGroup里完成，维护了线程池，大小如果不设置的话默认是2Ncpu，即CPU核数×2

EventLoopGroup的初始化过程。

（1）EventLoopGroup（其实是MultithreadEventExecutorGroup）内部维护一个类型为EventExecutor的children数组，其大小是nThreads，这样就构成了一个线程池。

（2）我们在实例化NioEventLoopGroup时，如果指定线程池大小，则nThreads就是指定的值，反之是CPU核数×2。

（3）在MultithreadEventExecutorGroup中调用newChild()象方法来初始化children数组。

（4）newChild()方法是在NioEventLoopGroup中实现的，它返回一个NioEventLoop实例。

（5）初始化NioEventLoop对象并给属性赋值，具体赋值的属性如下。

     ● provider：就是在NioEventLoopGroup构造器中，调用SelectorProvider.provider()方法获取的SelectorProvider对象。

       ● selector：就是在NioEventLoop构造器中，调用provider.openSelector()方法获取的Selector对象

题外话：线程池数量的选择上的一般规律：

Nthreads=Ncpu*Ucpu*(1+w/c)，其中

Ncpu=CPU核心数

Ucpu=cpu使用率，0~1

W/C=等待时间与计算时间的比率

Nthreads=Ncpu*(1+w/c)

IO密集型：一般情况下，如果存在IO，那么肯定w/c>1（阻塞耗时一般都是计算耗时的很多倍）,但是需要考虑系统内存有限（每开启一个线程都需要内存空间），这里需要上服务器测试具体多少个线程数适合（CPU占比、线程数、总耗时、内存消耗）。如果不想去测试，保守点取1即，Nthreads=Ncpu*(1+1)=2Ncpu。这样设置一般都OK。

计算密集型：假设没有等待w=0，则W/C=0. Nthreads=Ncpu。

至此结论就是：

IO密集型=2Ncpu（可以测试后自己控制大小，2Ncpu一般没问题）（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等）

计算密集型=Ncpu（常出现于线程中：复杂算法）

java中：Ncpu=Runtime.getRuntime().availableProcessors()

对于chooser的选择，使用的判断法，如果nThreads是2的平方，则使用PowerOfTwoEventExecutorChooser，否则使用GenericEventExecutorChooser。这里有趣的是判断2的倍数的方法，因为2倍数的相反数的反码的补码和源码相同，所以val & -val == val 就会证明了val是否是2的倍数

 public EventExecutorChooser newChooser(EventExecutor[] executors) { if (isPowerOfTwo(executors.length)) { return new PowerOfTwoEventExecutorChooser(executors); } else { return new GenericEventExecutorChooser(executors); } } private static boolean isPowerOfTwo(int val) { return (val & -val) == val; }

还有个有趣的事情就是chooser的两种方式的原因是，netty对next方法的优化，比如如下代码中的，AtomicInteger自增长取余的过程，如果是2的倍数，使用&会比%效率更高，因为位运算是直接在内存中进行，避免了10进制转成2进制到内存中进行计算，然后再把结果转换成10进制的过程。这一点对于很通用，比如hashmap的大小建议为2的n次方的原因，有比如负载均衡轮询的自增长取余过程，等等，当然还有个有趣的是，int无线增值的循环往复0-->2^31-->-2^31--0，因为计算机使用补码，可以将符号位和其它位统一处理。

 public EventExecutor next() { return executors[idx.getAndIncrement() & executors.length - 1]; } public EventExecutor next() { return executors[Math.abs(idx.getAndIncrement() % executors.length)]; }

• 将Channel注册到Selector

     Channel会在Bootstrap的connect的initAndRegister()中进行初始化，并且这个方法还会将初始化好的Channe注册到NioEventLoop的Selector中。接下来我们分析一下Channel注册的过程。

Channel的注册过程，具体如下。

 (1)在AbstractBootstrap的initAndRegister()方法中，通过group().register(channel)调用MultithreadEventLoopGroup的register()方法。

  （2）在MultithreadEventLoopGroup的register()方法中，调用next()方法获取一个可用的SingleThreadEventLoop，然后调用它的register()方法。

 (3)在SingleThreadEventLoop的register()方法中，调用channel.unsafe().register(this，promise)方法获取Channel的unsafe()底层操作对象，然后调用Unsafe的register()方法。

  （4）在AbstractUnsafe的register()方法中，调用register0()方法注册Channel对象。

  （5）在AbstractUnsafe的register0()方法中，调用AbstractNioChannel的doRegister()方法。

  (6)AbstractNioChannel的doRegister()方法通过javaChannel().register(eventLoop().selector，0，this)将Channel对应的Java NIO的SocketChannel注册到一个eventLoop的Selector中，并且将当前Channel作为Attachment与SocketChannel关联。

• connect过程如下图所示