Netty源码之Channel系列

今天药忘吃喽~ 2022-12-29 11:29 199阅读 0赞

# 源码之Channel系列 #

NioServerSocketChannel的继承关系图：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

NioSocketChannel的继承关系图：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

## Channel ##

JDK中的Channel是NIO类库的一个重要组成部分，是java.nio.SocketChannel和java.nio.ServerSocketChannel最顶层的接口，它们用于非阻塞的IO操作。

类似于NIO的Channel，Netty提供了自己的Channel和其子类实现，用于异步IO操作和其他相关的操作。

io.netty.channel.Channel是Netty网络操作抽象接口，同时也包含了Netty框架相关的一些功能，包括获取该Chanel的EventLoop，获取缓冲分配器ByteBufAllocator和pipeline等。

主要方法描述如下：

*  read()：从当前的Channel中读取数据到第一个inbound缓冲区中，如果数据被成功读取，触发ChannelHandler.channelRead(ChannelHandlerContext, Object)事件。读取操作API调用完成之后，紧接着会触发ChannelHandler.channelReadComplete(ChannelHandlerContext)事件，这样业务的ChannelHandler可以决定是否需要继续读取数据。
 *  flush()：将之前写入到发送缓冲区数组中的消息全部写入到目标Chanel中，发送给通信对方。
 *  config()：获取当前Channel的配置信息，例如CONNECT\_TIMEOUT\_MILLIS。
 *  isOpen()：判断当前Channel是否已经打开。
 *  isRegistered()：判断当前Channel是否已经注册到EventLoop上。
 *  isActive()：判断当前Channel是否已经处于激活状态。
 *  metadata()：获取当前Channel的元数据描述信息，包括TCP参数配置等。当创建Socket的时候需要指定TCP参数，例如接收和发送的TCP缓冲区大小、TCP的超时时间、是否重用地址等。在Netty中，每个Channel对应一个物理连接，每个连接都有自己的TCP参数配置。所以，Channel会聚合一个ChannelMetadata用来对TCP参数提供元数据描述信息，通过metadata()方法就可以获取当前Channel的TCP参数配置。
 *  localAddress()：获取当前Channel的本地绑定地址。
 *  remoteAddress()：获取当前Channel通信的远程Socket地址。
 *  eventLoop()：Channel需要注册到EventLoop的多路复用器上，用于处理I/O事件，通过eventLoop()方法可以获取到Channel注册的EventLoop。EventLoop本质上就是处理网络读写事件的Reactor线程。在Netty中，它不仅仅用来处理网络事件，也可以用来执行定时任务和用户自定义NioTask等任务。
 *  parent()：对于服务端Channe而言，它的父Channel为空；对于客户端 Channel，它的父Channel就是创建它的ServerSocketChannel。
 *  id()：它返回Channelld对象，Channelld是Channel的唯一标识。

## AbstractChannel ##

AbstractChannel是Netty网络操作抽象接口，它聚合了一组功能，包括但不限于网路的读、写，客户端发起连接，主动关闭连接，链路关闭，获取通信双方的网络地址等。

通过变量定义可以看出，AbstractChannel 聚合了所有 Channel 使用到的能力对象，由AbstractChannel 提供初始化和统一封装，如果功能和子类强相关，则定义成抽象方法由子类具体实现。

private final Channel parent;
    private final ChannelId id;
    private final Unsafe unsafe;
    private final DefaultChannelPipeline pipeline;
    private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
    private final CloseFuture closeFuture = new CloseFuture(this);
    
    private volatile SocketAddress localAddress;
    private volatile SocketAddress remoteAddress;
    private volatile EventLoop eventLoop;
    private volatile boolean registered;
    private boolean closeInitiated;
    private Throwable initialCloseCause;
    
    /** Cache for the string representation of this channel */
    private boolean strValActive;
    private String strVal;

Netty基于事件驱动，我们也可以理解为当Chnanel进行IO操作时会产生对应的IO事件，然后驱动事件在ChannelPipeline中传播，由对应的ChannelHandler对事件进行拦截和处理，不关心的事件可以直接忽略。所以  
AbstractChannel中网络I/O操作直接调用DefaultChannelPipeline的相关方法，由DefaultChannelPipeline中对应的ChannelHandler进行具体的逻辑处理。

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

## AbstractNioChannel ##

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

首先看看这个类的重要属性：

private final SelectableChannel ch;
    protected final int readInterestOp;
    volatile SelectionKey selectionKey;
    ... ...
    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和请求的通信地址信息。

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

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) { 
                logger.warn(
                            "Failed to close a partially initialized socket.", e2);
            }
    
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }

doRegister()：将Channel注册到Selector上。

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对哪几类网络事件感兴趣，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异常到上层进行统一处理。

doBeginRead()：注册感兴趣的事件。

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中，我们很快就会看到。

## AbstractNioByteChannel ##

从这个类的名字可以看到，这个类负责在网络上实际读写数据的 byte，只不过在实现上，这个类实现了具体的写行为，更底层的写、读、写文件等，比如doWriteBytes、doReadBytes、doWriteFileRegion操作则留给了具体的子类去实现。

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

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

doWrite()：

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，这个方法有两个分支，一个分支判断需要发送的消息是否是ByteBuf类型，另一个分支判断消息发送的是文件传输类型，两者的实现原理很相似，我们主要看ByteBuf类型。

首先判断需要发送的消息是否是ByteBuf类型，如果是，则进行强制类型转换，将其转换成ByteBuf类型，判断当前消息的可读字节数是否为0，如果为 ，说明该消息不可读，需要丢弃。从发送消息缓存中删除该消息，退出到外部循环继续处理其他的消息。

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;
    }

调用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)方法：

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);
        }
    }

看这个方法的名字我们可以知道，它的含义是发送消息缓存ChannelOutboundBuffer还有数据未完成时的处理，这个方法的参数接受一个writeSpinCount和0的比较值，无非就是true和fasle。

如果writeSpinCount小于0（writeSpinCount什么时候才会出现小于0呢？上面已经分析过，如果调用doWriteBytes发送消息，发送的字节数为0，则返回一个WRITE\_STATUS\_SNDBUF\_FULL=Integer.MAX\_VALUE值。Socket的Buffer已经写满，无法再继续发送数据）为true，调用setOpWrite()方法，在Selector上注册写标识。首先判断SelectionKey是否有效，然后检查Selector上是否注册了OP\_WRITE标识，如果没有则注册上，这样的话在selector下次轮询的时候，将再次执行写操作。

protected final void setOpWrite() { 
        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);
        }
    }

如果writeSpinCount大于等于 0（大于0一般不可能，因为上面退出的循环条件是writeSpinCount>0），则清除Selector上注册的写标识，并且启动独立的Runnable，将其加入到EventLoop中执行，Runnable里的实现很简单，就是调用flush()方法将数据强制刷出。

## 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);
        }
    }
    
    private final ServerSocketChannelConfig config;

首先创建了静态的ChannelMetadata成员变量，然后定义了ServerSocketChannelConfig用于配置ServerSocketChannel的TCP参数。静态的newSocket方法用于通过ServerSocketChannel（JDK NIO中的类）的open打开新的ServerSocketChannel通道。

构造方法：

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

调用了父类的父类（AbstractNioChannel）的构造方法，将NioServerSocketChannel的所应该关注的事件设置为OP\_ACCEPT。

doBind()：

protected void doBind(SocketAddress localAddress) throws Exception { 
        if (PlatformDependent.javaVersion() >= 7) { 
            javaChannel().bind(localAddress, config.getBacklog());
        } else { 
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

javaChannel的实现是java.nio.ServerSocketChannel。doBind中服务端在进行端口绑定的时候，进行了JDK版本的判定，并且可以指定backlog，也就是允许客户端排队的最大长度。

doReadMessages(List buf)方法在父类AbstractNioMessageChannel进行了抽象定义，在这个类中有具体的实现，用来处理客户端的连接。

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 buf中，最后返回1，表示服务端消息读取成功。很明显，对于 NioServerSocketChannel，它的读取操作就是接收客户端的连接，创建NioSocketChannel对象。

## NioSocketChannel ##

NioSocketChannel比较重要，因为具体的网络读写都是在NioSocketChannel发生的。

构造方法：

public NioSocketChannel(Channel parent, SocketChannel socket) { 
        super(parent, socket);
        config = new NioSocketChannelConfig(this, socket.socket());
    }

NioSocketChannel是由上面NioServerSocketChannel中创建的，所以属性parent为NioServerSocketChannel。

doConnect()：

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连接。之后对连接结果进行判断，连接结果有以下三种可能。

1.  连接成功，返回true。
2.  暂时没有连接上，服务端没有返回ACK应答，连接结果不确定，返回false，需要将NioSocketChannel中的selectionKey设置为OP\_CONNECT，监听连接网络操作位。
3.  连接失败，直接抛出I/O异常。说明客户端的TCP握手请求直接被REST或者被拒绝，此时需要关闭客户端连接。

doReadBytes()和doWriteBytes()：处理数据的读和写。

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);
    }

总的来说，是基于NIO的SocketChannel和Netty的ByteBuf封装而成。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/fcd7eca4feb54d06b939a41b3064c778.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/803f0519d98547bd8cd7984d3201acfd.png