Channel 接口实现类系列源码之NioServerSocketChannel类分析（二十八）

拼搏现实的明天。 2022-10-29 14:21 170阅读 0赞

Channel 接口实现类系列源码之NioServerSocketChannel类分析

**一、主要类关系图**

***1、NioServerSocketChannel***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

***2、NioSocketChannel***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**二、Channel**  **接口**

Channel  是  JDK  的  NIO  类库的一个重要组成部分，就是  java.nio.SocketChannel  和java.nio.ServerSocketChannel，它们用于非阻塞的  IO  操作。 类似于 NIO  的  Channel ， Netty  提供了自己的  Channel  和其子类实现，用于异步  IO  操作和其他相关的操作。 io.netty.channel.Channel 是  Netty  网络操作抽象接口，同时也包含了  Netty  框架相关的一些功能，包括获取该 Chanel  的  EventLoop ，获取缓冲分配器  ByteBufAllocator  和  pipeline  等。如下类方法图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

主要方法描述如下：

1、Channel read(): 从当前的  Channel  中读取数据到第一个  inbound  缓冲区中，如果数据被成功读取，触发 ChannelHandler.channelRead(ChannelHandlerContext, Object) 事件。读取操作

API  调用完成之后 , 紧接着会触发  ChannelHandler.channelReadComplete (ChannelHandlerContext)事件，这样业务的  ChannelHandler  可以决定是否需要继续读取数据。

2、Channel flush(): 将之前写入到发送环形数组中的消息全部写入到目标  Chanel  中，发送给通信对方。

3、ChannelConfig config(): 获取当前  Channel  的配置信息，例如  CONNECT\_TIMEOUT\_MILLIS 。

4、boolean isOpen(): 判断当前  Channel  是否已经打开。

5、boolean isRegistered(): 判断当前  Channel  是否已经注册到  EventLoop  上。

6、boolean isActive(): 判断当前  Channel  是否已经处于激活状态。

7、ChannelMetadata metadata(): 获取当前  Channel  的元数据描述信息，包括  TCP  参数配置等。

8、SocketAddress localAddress(): 获取当前  Channel  的本地绑定地址。

9、SocketAddress remoteAddress(): 获取当前  Channel  通信的远程  Socket  地址。

10、EventLoop eventLoop() ： Channel  需要注册到  EventLoop  的多路复用器上，用于处理  I/O 事件，通过 eventLoop() 方法可以获取到  Channel  注册的  EventLoop 。 EventLoop  本质上就是

处理网络读写事件的  Reactor  线程。在  Netty  中，它不仅仅用来处理网络事件，也可以用来执行定时任务和用户自定义 NioTask  等任务。

11、ChannelMetadata metadata() ：当创建  Socket  的时候需要指定  TCP  参数，例如接收和发送的 TCP  缓冲区大小、 TCP  的超时时间、是否重用地址等。在  Netty  中，每个  Channel  对应

一个物理连接，每个连接都有自己的  TCP  参数配置。所以， Channel  会聚合一个 ChannelMetadata 用来对  TCP  参数提供元数据描述信息，通过  metadata() 方法就可以获取当 前 Channel  的  TCP  参数配置。

12、Channel parent() ：对于服务端  Channel  而言，它的父  Channel  为空 ; 对于客户端  Channel ， 它的父 Channel  就是创建它的  ServerSocketChannel 。

13、ChannelId id() ：它返回  Channelld  对象， Channelld  是  Channel  的唯一标识。

由类关系图可以看出， AbstractChannel  和  AbstractNioChannel  是实现类 NioServerSocketChannel 和  NioSocketChannel  的共同父类，所以这两个父类我们有必要了解一下。

**三、AbstractChannel**

AbstractChannel  是  Netty  网络操作抽象接口，它聚合了一组功能，包括但不限于网路的读、写，客户端发起连接，主动关闭连接，链路关闭，获取通信双方的网络地址等。 通过变量定义可以看出，AbstractChannel  聚合了所有  Channel  使用到的能力对象，由 AbstractChannel 提供初始化和统一封装，如果功能和子类强相关，则定义成抽象方法由子类 具体实现 Netty 基于事件驱动，我们也可以理解为当 Chnanel  进行  IO  操作时会产生对应的  IO  事件，然后驱动事件在 ChannelPipeline  中传播，由对应的  ChannelHandler  对事件进行拦截和处理，不关心的事件可以直接忽略。

AbstractChannel  中网络  I/O  操作直接调用  DefaultChannelPipeline  的相关方法，由 DefaultChannelPipeline 中对应的  ChannelHandler  进行具体的逻辑处理。

@Override
        public ChannelFuture bind(SocketAddress localAddress) {
            return pipeline.bind(localAddress);
        }
    
        @Override
        public ChannelFuture connect(SocketAddress remoteAddress) {
            return pipeline.connect(remoteAddress);
        }
    
        @Override
        public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
            return pipeline.connect(remoteAddress, localAddress);
        }
    
        @Override
        public ChannelFuture disconnect() {
            return pipeline.disconnect();
        }
    
        @Override
        public ChannelFuture close() {
            return pipeline.close();
        }
    
        @Override
        public ChannelFuture deregister() {
            return pipeline.deregister();
        }
    
        @Override
        public Channel flush() {
            pipeline.flush();
            return this;
        }

AbstractChannel 也提供了一些公共 API 的具体实现，例如 localAddress()和 remoteAddress()方法：

@Override
    public SocketAddress remoteAddress() {
        SocketAddress remoteAddress = this.remoteAddress;
        if (remoteAddress == null) {
            try {
                this.remoteAddress = remoteAddress = unsafe().remoteAddress();
            } catch (Throwable t) {
                // Sometimes fails on a closed socket in Windows.
                return null;
            }
        }
        return remoteAddress;
    }

首先从成员变量中获取，如果第一次调用为空，需要通过 unsafe 的 remoteAddress 获取， 它是个抽象方法，具体由对应的 Channel 子类实现。 从这个 AbstractChannel 来看，Unsafe 系列的类非常重要，我们将在后面说明。

**四、AbstractNioChannel**

顾名思义  AbstractNioChanne  是  Netty  对  NIO  类型网络操作的抽象，所以其中的成员变量和方法必然和 NIO  网络通信高度相关。

***1、成员变量***

首先看看这个类的重要成员变量

private static final InternalLogger logger =
            InternalLoggerFactory.getInstance(AbstractNioChannel.class);
    
    private static final ClosedChannelException DO_CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractNioChannel.class, "doClose()");
    
    private final SelectableChannel ch;
    protected final int readInterestOp;
    volatile SelectionKey selectionKey;
    boolean readPending;
    private final Runnable clearReadPendingRunnable = new Runnable() {
        @Override
        public void run() {
            clearReadPending0();
        }
    };
    
    /**
     * The future of the current connection attempt.  If not null, subsequent
     * connection attempts will fail.
     */
    private ChannelPromise connectPromise;
    private ScheduledFuture<?> connectTimeoutFuture;
    private SocketAddress requestedRemoteAddress;

由于 NioSocketChannel 和 NioServerSocketChannel 需要共用，所以定义了一个 java.nio.SocketChannel 和 java.nio.ServerSocketChannel 的公共父类 SelectableChannel（由 JDK 定义），用于设置 SelectableChannel 参数和进行 IO 操作。 第二个参数是 readInterestOp，表示当前的 Channel 所关注的事件，不过从名字上来说， 我们可以认为这个 AbstractNioChannel 关注的事件是读。 随后定义了一个 volatile 修饰的 SelectionKey，该 SelectionKey 是 Channel 注册到 EventLoop 后返回的选择键。由于 Channel 会面临多个业务线程的并发写操作，当SelectionKey 由 SelectionKey 修改之后，为了能让其他业务线程感知到变化，所以需要使用 volatile 保证修改的可见性。 很明显，上述的成员变量都是和 NIO 操作相关的。 最后定义了代表连接操作结果的 ChannelPromise 以及连接超时定时器 ScheduledFuture 和请求的通信地址信息。

***2、主要***  ***API***

构造方法

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false);
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
                if (logger.isWarnEnabled()) {
                    logger.warn(
                            "Failed to close a partially initialized socket.", e2);
                }
            }
    
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }

这个类的构造方法的用意很明显，让 AbstractNioChanne 中的成员变量实例化，并且设置 SelectableChannel 为非阻塞模式。

3、doRegister

@Override
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                if (!selected) {
                    // Force the Selector to select now as the "canceled" SelectionKey may still be
                    // cached and not removed because no Select.select(..) operation was called yet.
                    eventLoop().selectNow();
                    selected = true;
                } else {
                    // We forced a select operation on the selector before but the SelectionKey is still cached
                    // for whatever reason. JDK bug ?
                    throw e;
                }
            }
        }
    }

首先是 Channel 的注册，定义一个布尔类型的局部变量 selected 来标识注册操作是否成功，调用 SelectableChannel 的 register 方法，将当前的 Channel 注册到 EventLoop 的多路 复用器上。 注册 Channel 的时候需要指定监听的网络操作位来表示 Channel 对哪几类网络事件感兴趣，前面我们说过在向 Selector 对象注册感兴趣的事件时。

JAVA NIO  共定义了四种： OP\_READ （1 ）、 OP\_WRITE （ 4 ）、 OP\_CONNECT （ 8 ）、 OP\_ACCEPT （ 16 ），都定义在  SelectionKey 中，分别对应读、写、请求连接、接受连接等网络 Socket  操作。 AbstractNioChannel 注册的是  0 ，说明对任何事件都不感兴趣，仅仅完成注册操作。注册的时候可以指定附件，后续 Channel  接收到网络事件通知时可以从  SelectionKey  中重新获 取之前的附件进行处理，此处将 AbstractNioChannel  的实现子类自身当作附件注册。如果注册 Channel  成功，则返回  selectionKey ，通过  selectionKey  可以从多路复用器中获取  Channel 对象。如果当前注册返回的 selectionKey  已经被取消，则抛出  CancelledKeyException  异常，捕获该异常进行处理。如果是第一次处理该异常，调用多路复用器的 selectNow() 方法将已经取消的 selectionKey  从多路复用器中删除掉。操作成功之后，将  selected  置为  true ，说明之前失效的 selectionKey  已经被删除掉。继续发起下一次注册操作，如果成功则退出，如果仍然发生 CancelledKeyException  异常 , 说明我们无法删除已经被取消的  selectionKey, 按照  JDK  的API 说明，这种意外不应该发生。如果发生这种问题，则说明可能  NIO  的相关类库存在不可恢复的 BUG ，直接抛出  CancelledKeyException  异常到上层进行统一处理。

4、doBeginRead

@Override
    protected void doBeginRead() throws Exception {
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }
    
        readPending = true;
    
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

这个方法的作用是在准备处理读操作之前需要设置网路操作位为读。 先判断下 Channel 是否关闭，如果处于关闭中，则直接返回。获取当前的 SelectionKey 进行判断，如果可用，说明 Channel 当前状态正常，则可以进行正常的操作位修改。 将 SelectionKey 当前的操作位与读操作位进行按位与操作，如果等于 0，说明目前并没有设置读操作位，通过 interestOps | readInterestOp 设置读操作位，最后调用 selectionKey 的 interestOps 方法重新设置通道的网络操作位，这样就可以监听网络的读事件了。 回顾一下，readInterestOp 是在哪里被设置为读相关的呢？在 AbstractNioChannel 的构 造方法里，由参数传入，这个构造方法的调用，则是在 AbstractNioChannel 的子类AbstractNioByteChannel 中，我们很快就会看到。

![20210215162040741.png][]

**五、AbstractNioByteChannel**

成员变量很少，只有一个  Runnable  类型的  flushTask  来负责继续写半包消息。 从这个类的名字可以看到，这个类负责在网络上实际读写数据的 byte ，只不过在实现上，这个类实现了具体的写行为，更底层的写、读、写文件等，比如doWriteBytes 、doReadBytes、doWriteFileRegion 操作则留给了具体的子类去实现。

***1、主要***  ***API***

构造方法

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ);
    }

这个方法里调用了父类 AbstractNioChannel 的构造方法，并且设置关注的事件为读事件 （SelectionKey.OP\_READ）

2、doWrite （流程图 ? ）

@Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        int writeSpinCount = config().getWriteSpinCount();
        do {
            Object msg = in.current();
            if (msg == null) {
                // Wrote all messages.
                clearOpWrite();
                // Directly return here so incompleteWrite(...) is not called.
                return;
            }
            writeSpinCount -= doWriteInternal(in, msg);
        } while (writeSpinCount > 0);
    
        incompleteWrite(writeSpinCount < 0);
    }

首先设置了一个变量 writeSpinCount，表示写操作的最大循环计数，writeSpinCount 缺省为 16 次（设置最大循环次数的原因是当循环发送的时候，IO 线程会一直尝试进行写操作， 此时 I/O 线程无法处理其他的 I/O 操作，例如读新的消息或者执行定时任务和 NioTask 等，如果网络 I/O 阻塞或者对方接收消息太慢，可能会导致线程假死）。 从发送消息缓存 ChannelOutboundBuffer 弹出一条消息，判断该消息是否为空，如果为空，说明发送消息缓存 ChannelOutboundBuffer 中所有待发送的消息都已经发送完成，清除写操作在 selector 上的注册，然后退出循环。清除写操作注册的 clearOpWrite 方法实现如下：

protected final void clearOpWrite() {
        final SelectionKey key = selectionKey();
        // Check first if the key is still valid as it may be canceled as part of the deregistration
        // from the EventLoop
        // See https://github.com/netty/netty/issues/2104
        if (!key.isValid()) {
            return;
        }
        final int interestOps = key.interestOps();
        if ((interestOps & SelectionKey.OP_WRITE) != 0) {
            key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
        }
    }

从当前 SelectionKey 中获取网络操作位，然后与 SelectionKey.OP\_WRITE 做按位与，如果不等于 0，说明当前的 SelectionKey 是 isWritable 的，需要清除写操作位。清除方法很简单，就是 SelectionKey.OP\_WRITE 取非之后与原操作位做按位与操作，清除 SelectionKey 的写操作位。如果需要发送的消息不为空，则继续处理，具体的处理方法是 doWriteInternal，

private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) msg;
            if (!buf.isReadable()) {
                in.remove();
                return 0;
            }
    
            final int localFlushedAmount = doWriteBytes(buf);
            if (localFlushedAmount > 0) {
                in.progress(localFlushedAmount);
                if (!buf.isReadable()) {
                    in.remove();
                }
                return 1;
            }
        } else if (msg instanceof FileRegion) {
            FileRegion region = (FileRegion) msg;
            if (region.transferred() >= region.count()) {
                in.remove();
                return 0;
            }
    
            long localFlushedAmount = doWriteFileRegion(region);
            if (localFlushedAmount > 0) {
                in.progress(localFlushedAmount);
                if (region.transferred() >= region.count()) {
                    in.remove();
                }
                return 1;
            }
        } else {
            // Should not reach here.
            throw new Error();
        }
        return WRITE_STATUS_SNDBUF_FULL;
    }

这个方法有两个分支，一个分支判断需要发送的消息是否是 ByteBuf 类型，另一个分支判断消息发送的是文件传输类型，两者的实现原理很相似，我们主要看 ByteBuf 类型。 首先判断需要发送的消息是否是 ByteBuf 类型，如果是，则进行强制类型转换，将其转换成 ByteBuf 类型，判断当前消息的可读字节数是否为 0，如果为 0，说明该消息不可读， 需要丢弃。从发送消息缓存中删除该消息，退出到外部循环继续处理其他的消息。调用 doWriteBytes 进行消息发送，不同的 Channel 子类有不同的实现，因此它是抽象方法，但其实也只有两个实现，分别对应着 TCP 和 UDP。 如果发送的字节数大于 0，更新发送消息缓存 ChannelOutboundBuffer 里的发送进度信息，同时判断当前消息的可读字节数是否为 0，如果为 0，说明该消息已经处理完成，从发送消息缓存 ChannelOutboundBuffer 中删除该消息，退出到外部循环继续处理其他的消息。如果调用 doWriteBytes 发送消息，发送的字节数为 0，则返回一个 WRITE\_STATUS\_SNDBUF\_FULL = Integer.MAX\_VALUE 值。 一般只有当前 Socket 缓冲区写满了，无法再继续发送数据的时候才会返回0（Socket 的 Buffer 已满）。 如果继续循环发送也还是无法写入的，这时只见返回一个比较大值，会直接退出循环发送的，稍后再尝试写入。方法的返回并且返回值表示本次写操作成功的次数。回到 doWrite 方法，发送消息缓存 ChannelOutboundBuffer 没写完，需要循环继续处理，根据 doWriteInternal 方法的返回值对写操作的循环计数器 writeSpinCount 进行扣减，如果没 有超过最大循环计数，继续从发送消息缓存 ChannelOutboundBuffer 中读取数据进行处理。接下来调用 incompleteWrite(writeSpinCount < 0);方法，看这个方法的名字我们可以知道， 它的含义是发送消息缓存 ChannelOutboundBuffer 还有数据未完成时的处理，这个方法的参数接受一个 writeSpinCount 和 0 的比较值，无非就是 true 和 fasle。

protected final void incompleteWrite(boolean setOpWrite) {
        // Did not write completely.
        if (setOpWrite) {
            setOpWrite();
        } else {
            // It is possible that we have set the write OP, woken up by NIO because the socket is writable, and then
            // use our write quantum. In this case we no longer want to set the write OP because the socket is still
            // writable (as far as we know). We will find out next time we attempt to write if the socket is writable
            // and set the write OP if necessary.
            clearOpWrite();
    
            // Schedule flush again later so other tasks can be picked up in the meantime
            eventLoop().execute(flushTask);
        }
    }

如果 writeSpinCount 小于 0（writeSpinCount 什么时候才会出现小于 0 呢？上面已经分析过，如果调用 doWriteBytes 发送消息，发送的字节数为 0，则返回一个 WRITE\_STATUS\_SNDBUF\_FULL = Integer.MAX\_VALUE 值。Socket 的 Buffer 已经写满，无法再继续发送数据），为 true，调用 setOpWrite() 方法，在 Selector 上注册写标识。在首先判断 SelectionKey 是否有效，然后检查 Selector 上是否注册了 OP\_WRITE 标识，如果没有则注册上，这样的话在 selecto 下次轮询的时候，将再次执行写操作。 如果 writeSpinCount 大于等于 0（大于 0 一般不可能，因为上面退出的循环条件是 writeSpinCount >0），则清除 Selector 上注册的写标识，并且启动独立的 Runnable，将其加入到 EventLoop 中执行，Runnable 里的实现很简单，就是调用 flush()方法将数据强制刷出。 这段代码意义何在？这是 Netty 的机制之一，作用是让每个连接默认最多连续写 16 次 （writeSpinCount 的缺省值），超过 16 次，即使还有数据以及缓冲区还有空闲也暂时不处理了，先处理下一个连接。这个 Channel 上没有写完的数据，就直接 schedule 一个 task 来 继续写，空闲时继续刷新，而不是用注册写事件来触发，推测 Netty 的设计思路是，这个Channel 可能只是因为自旋次数少，引起的数据发送不完全，而无需等待 selector，干脆就 在空闲时继续处理这个 Channel 上的数据。

**六、AbstractNioMessageChannel**

AbstractNioMessageChannel  代码比较简单，没有成员变量，构造方法里只调用了的父类AbstractNioChannel 的构造方法。 关键的方法是 doWrite  和  doBeginRead ， doWrite  的设计思路和  AbstractNioByteChannel 中的 doWrite  方法差不太多。而且对于  AbstractNioMessageChannel  的在  NIO  下的实现类 NioServerSocketChannel 来说，这个方法也没有什么意义，所以不再重复讲述。 doBeginRead 方法则是直接调用的父类  AbstractNioChannel  的  doBeginRead  方法。

**七、NioServerSocketChannel**

NioServerSocketChannel  在功能上主要是接受客户端的连接，所以在实现上比较简单。

private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NioServerSocketChannel.class);
    
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            /**
             *  Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in
             *  {@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise.
             *
             *  See <a href="https://github.com/netty/netty/issues/2308">#2308</a>.
             */
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }

首先创建了静态的 ChannelMetadata 成员变量,然后定义了 ServerSocketChannelConfig 用于配置 ServerSocketChannel 的 TCP 参数。静态的 newSocket 方法用于通过

ServerSocketChannel  的  open  打开新的  ServerSocketChannel  通道。

***1、*** ***构造方法***

public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

调用了父类的父类（AbstractNioChannel）的构造方法，将 NioServerSocketChannel 的所应该关注的事件设置为 OP\_ACCEPT。当然，从前面我们对 AbstractNioChannel 的解读我们看到，进行实际的设置是在 doBeginRead 方法中。

***2、主要***  ***API***

接着我们再看下  NioServerSocketChannel  相关的接口实现 : isActive 、 remoteAddress 、 javaChannel 和  doBind：

@Override
    public boolean isActive() {
        return javaChannel().socket().isBound();
    }
    
    @Override
    public InetSocketAddress remoteAddress() {
        return null;
    }
    
    @Override
    protected ServerSocketChannel javaChannel() {
        return (ServerSocketChannel) super.javaChannel();
    }
    
    @Override
    protected SocketAddress localAddress0() {
        return SocketUtils.localSocketAddress(javaChannel().socket());
    }
    
    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

这几个方法都很简单，isActive 通过 java.net.ServerSocket 的 isBound 方法判断服务端监听端口是否处于绑定状态，它的 remoteAddress 为空。javaChannel 的实现是 java.nio.ServerSocketChannel。doBind 中服务端在进行端口绑定的时候，进行了 JDK 版本的判定，并且可以指定 backlog，也就是允许客户端排队的最大长度。 doReadMessages(List<Object> buf)方法在父类 AbstractNioMessageChannel 进行了抽象定义，在这个类中有具体的实现，用来处理客户端的连接

@Override
    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = SocketUtils.accept(javaChannel());
    
        try {
            if (ch != null) {
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            logger.warn("Failed to create a new channel from an accepted socket.", t);
    
            try {
                ch.close();
            } catch (Throwable t2) {
                logger.warn("Failed to close a socket.", t2);
            }
        }
    
        return 0;
    }

首先通过 SocketUtils 的工具类 的 accept 接收新的客户端连接，如果 SocketChannel 不 为空，则创建新的 NioSocketChannel，并将其加入到 List<Object> buf 中，最后返回 1，表示 服务端消息读取成功。很明显，对于 NioServerSocketChannel，它的读取操作就是接收客户端的连接，创建 NioSocketChannel 对象。 至于其他与服务端 Channel 无关的接口定义，由于这些方法是客户端 Channel 相关的，因此，对于服务端 Channel 无须实现。如果这些方法被误调，则返回UnsupportedOperationException 异常。

@Override
    protected boolean doConnect(
            SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        throw new UnsupportedOperationException();
    }
    
    @Override
    protected void doFinishConnect() throws Exception {
        throw new UnsupportedOperationException();
    }

八、NioSocketChannel
    
    NioSocketChannel 比较重要，因为具体的网络读写都是在 NioSocketChannel 发生的。 
    
    1、连接方法 doConnect

@Override
    protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        if (localAddress != null) {
            doBind0(localAddress);
        }
    
        boolean success = false;
        try {
            boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
            if (!connected) {
                selectionKey().interestOps(SelectionKey.OP_CONNECT);
            }
            success = true;
            return connected;
        } finally {
            if (!success) {
                doClose();
            }
        }
    }

判断本地 Socket 地址是否为空，如果不为空则调用 doBind0 方法，当然本质上是调用 JDK 的相关方法绑定本地地址。接下来，则继续调用
    
    java.nio.channels.SocketChannel.connect(SocketAddress remote)发起 TCP 连接。
    
    2、之后对连接结果进行判断，连接结果有以下三种可能。

(1 ）连接成功，返回  true;

(2 ）暂时没有连接上，服务端没有返回  ACK  应答，连接结果不确定，返回  false;

(3 ）连接失败，直接抛出  I/O  异常。

如果是结果 (2) ，需要将  NioSocketChannel  中的  selectionKey  设置为  OP\_CONNECT, 监听连接网络操作位。

如果抛出了  IO  异常，说明客户端的  TCP  握手请求直接被  REST  或者被拒绝，此时需要关闭客户端连接，。

***3、写方法***  ***doWrite***

对父类的  doWrite  方法进行了覆盖  Override ，它的实现比较复杂，在整体上，依然类似于 AbstractNioByteChannel  中  doWrite  方法的实现，有一个写操作的最大循环计数，最多连

续写最大次数，超过这个最大循环次数，即使还有数据也暂时不处理了。

具体代码:

@Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        SocketChannel ch = javaChannel();
        int writeSpinCount = config().getWriteSpinCount();
        do {
            if (in.isEmpty()) {
            //首先对发送消息缓存 ChannelOutboundBuffer 检查，如果为空，说明发送消息缓存 ChannelOutboundBuffer 
            //中所有待发送的消息都已经发送完成，清除写操作在 selector 上的注册，然后退出本方法。
                // All written so clear OP_WRITE
                clearOpWrite();
                // Directly return here so incompleteWrite(...) is not called.
                return;
            }
    
            // Ensure the pending writes are made of ByteBufs only.
             //接下来获取 ChannelOutboundBuffer 中 ByteBuf 的数量，在获取的时候会通过（1024, maxBytesPerGatheringWrite）
            //控制获取的 ByteBuf 的最大数量和最大字节数。 根据获取的 ByteBuf 的数量分情况进行处理：
            int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();
            ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);
            int nioBufferCnt = in.nioBufferCount();
    
            // Always us nioBuffers() to workaround data-corruption.
            // See https://github.com/netty/netty/issues/2761
            switch (nioBufferCnt) {
                case 0:
             
           //源码上注释的意思是：除了 ByteBuffers 之外，我们还有其他要写的东西，因此可以回退到普通写操作，它的意思是如果 nioBuffers 个数为 0，
             //可能是其他内容的数据，比如 FileRegion，则调用父类 AbstractNioByteChannel 的 doWriter0 方法去写。
    
                    // We have something else beside ByteBuffers to write so fallback to normal writes.
                    writeSpinCount -= doWrite0(in);
                    break;
                case 1: {
                
                //表示如果只有一个 ByteBuffer，则不需要使用 nio 的 gathering write 去写，而是直接使 用 SocketChannel 的 write 方法，
                //如果写入字节数小于等于 0，就直接注册写事件，退出方法， 等待下次 select 时，继续写。 如果写入了一定量的数据，
                // 则调整最大字节数，并将已经写入的数据从发送消息缓存 ChannelOutboundBuffer 中移除，然后开始新的一轮循环写。 
                    // Only one ByteBuf so use non-gathering write
                    // Zero length buffers are not added to nioBuffers by ChannelOutboundBuffer, so there is no need
                    // to check if the total size of all the buffers is non-zero.
                    ByteBuffer buffer = nioBuffers[0];
                    int attemptedBytes = buffer.remaining();
                    final int localWrittenBytes = ch.write(buffer);
                    if (localWrittenBytes <= 0) {
                        incompleteWrite(true);
                        return;
                    }
                    adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite);
                    in.removeBytes(localWrittenBytes);
                    --writeSpinCount;
                    break;
                }
                default: {
                 //如果 ByteBuf 个数大于 1，则使用 Nio 的 gather 聚合 Buffer 写入，一次就可以写入多个 Buffer 的数据。在实现上，和只有一个 ByteBuffer 时没有太大差别。
    
                    // Zero length buffers are not added to nioBuffers by ChannelOutboundBuffer, so there is no need
                    // to check if the total size of all the buffers is non-zero.
                    // We limit the max amount to int above so cast is safe
                    long attemptedBytes = in.nioBufferSize();
                    final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
                    if (localWrittenBytes <= 0) {
                        incompleteWrite(true);
                        return;
                    }
                    // Casting to int is safe because we limit the total amount of data in the nioBuffers to int above.
                    adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes,
                            maxBytesPerGatheringWrite);
                    in.removeBytes(localWrittenBytes);
                    --writeSpinCount;
                    break;
                }
            }
        } while (writeSpinCount > 0);
    
        incompleteWrite(writeSpinCount < 0);
    }

4、  *NIO*  *知识点*  *Gather*  *和*  *Scatter*

*聚集（* *gather* *）写入*  *Channel*  *是指在写操作时将多个*  *buffer*  *的数据写入同一个*  *Channel* *，*  *因此，* *Channel*  *将多个*  *Buffer*  *中的数据“聚集（* *gather* *）”后发送到*  *Channel* *。*

*分散（* *scatter* *）从*  *Channel*  *中读取是指在读操作时将读取的数据写入多个*  *buffer*  *中。因* *此，* *Channel*  *将从*  *Channel*  *中读取的数据“分散（* *scatter* *）”到多个*  *Buffer*  *中。*

***5、读写操作***

更底层的写、读、写文件等，比如  doWriteBytes 、 doReadBytes 、 doWriteFileRegion  操作，在 NioSocketChannel  的父类  AbstractNioByteChannel  中是抽象方法，所以在这个类中有自己的实现。

@Override
    protected int doReadBytes(ByteBuf byteBuf) throws Exception {
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.attemptedBytesRead(byteBuf.writableBytes());
        return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
    }
    
    @Override
    protected int doWriteBytes(ByteBuf buf) throws Exception {
        final int expectedWrittenBytes = buf.readableBytes();
        return buf.readBytes(javaChannel(), expectedWrittenBytes);
    }
    
    @Override
    protected long doWriteFileRegion(FileRegion region) throws Exception {
        final long position = region.transferred();
        return region.transferTo(javaChannel(), position);
    }

总的来说，是基于 NIO 的 SocketChannel 和 Netty 的 ByteBuf 封装而成，ByteBuf 的源码分析，我们后面会讲到。 唯一需要注意的是，在写文件块 doWriteFileRegion 方法中，使用了 JDK 提供的零拷贝技术

Channel 接口实现类系列源码之NioServerSocketChannel类分析完成，下篇分析Unsafe 接口实现类相关源码，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/c68a3273fcb44c4fa8dbf9fe31c4311c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/969253608d3c44659b8c7581198b68fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/eeb2268a28304639b9cb50efa6053c13.png
[20210215162040741.png]: /images/20221024/b5448634515649beb1880f3f8e8abf1f.png