Unsafe 接口实现类相关源码分析（二十九）

﹏ヽ暗。殇╰゛Y 2022-10-29 14:22 190阅读 0赞

今天对Unsafe 接口实现类相关源码分析：

简介：

Unsafe  是  Channel  的内部接口，聚合在  Channel  中协助进行网络读写相关的操作，因为它的设计初衷就是 Channel  的内部辅助类，不应该被  Netty  框架的上层使用者调用，所以被

命名为  Unsafe 。这里不能仅从字面理解认为它是不安全的操作，而要从整个架构的设计层面体会它的设计初衷和职责。Unsafe 用于处理 Channel  对应网络  IO  的底层操作。 ChannelHandler  处理回调事件时产生的相关网络 IO  操作最终也会委托给  Unsafe  执行。所以基本上一个  Channel  的实现中就有一个对应的 Unsafe  类的实现。

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

从类的层次关系上来说，和  Channel  系列非常的相近。

***1、NioMessageUnsafe***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

***2、NioSocketChannelUnsafe***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**3、Unsafe 类方法**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

**4、NioUnsafe** **类方法**

![20210215163236897.png][]

Unsafe  接口中定义了  socket  相关操作，包括  SocketAddress  获取、 selector  注册、网卡端口绑定、socket  建连与断连、 socket  写数据。这些操作都和  jdk  底层  socket  相关。

NioUnsafe  接口主要增加了一些和  nio  操作联系比较紧密的操作。

**二、AbstractUnsafe**

可以看见  AbstractUnsafe  中的方法很多，我们选择其中比较重要的方法进行研究。

***1、方法***  ***register***

register  方法主要用于将当前  Unsafe  对应的 Channel  注册到  EventLoop  的多路复用器上，然后调用 DefaultChannelPipeline  的  fireChannelRegistered  方法。如果  Channel  被激活，则调用 DefaultChannelPipeline  的  fireChannelActive  方法。

@Override
    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        if (eventLoop == null) {
            throw new NullPointerException("eventLoop");
        }
        if (isRegistered()) {
            promise.setFailure(new IllegalStateException("registered to an event loop already"));
            return;
        }
        if (!isCompatible(eventLoop)) {
            promise.setFailure(
                    new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
            return;
        }
    
        AbstractChannel.this.eventLoop = eventLoop;
    
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                logger.warn(
                        "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                        AbstractChannel.this, t);
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }
    }

首先判断当前所在的线程是否是 Channel 对应的 NioEventLoop 线程，如果是同一个线程，则不存在多线程并发操作问题，直接调用 register0 方法进行注册;如果是由用户线程或 者其他线程发起的注册操作，则将注册操作封装成 Runnable，放到 NioEventLoop 任务队列 中执行。注意:如果直接执行 register0 方法，会存在多线程并发操作 Channel 的问题。 接下来看 register0 方法的实现

private void register0(ChannelPromise promise) {
        try {
            // check if the channel is still open as it could be closed in the mean time when the register
            // call was outside of the eventLoop
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            boolean firstRegistration = neverRegistered;
            doRegister();
            neverRegistered = false;
            registered = true;
    
            // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
            // user may already fire events through the pipeline in the ChannelFutureListener.
            pipeline.invokeHandlerAddedIfNeeded();
    
            safeSetSuccess(promise);
            pipeline.fireChannelRegistered();
            // Only fire a channelActive if the channel has never been registered. This prevents firing
            // multiple channel actives if the channel is deregistered and re-registered.
            if (isActive()) {
                if (firstRegistration) {
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    // This channel was registered before and autoRead() is set. This means we need to begin read
                    // again so that we process inbound data.
                    //
                    // See https://github.com/netty/netty/issues/4805
                    beginRead();
                }
            }
        } catch (Throwable t) {
            // Close the channel directly to avoid FD leak.
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

首先调用 ensureOpen 方法判断当前 Channel 是否打开，如果没有打开则无法注册，直 接返回。校验通过后调用 doRegister 方法,它由 AbstractNioUnsafe 对应的 AbstractNioChannel 实现，前面已经讲述过实现。然后，调用 pipeline.invokeHandlerAddedIfNeeded()，保证在 Channel 注册之后立刻回调 handlerAdded，以保证 handler 被添加到 pipeline 上。 调用 pipeline 的 fireChannelRegistered 触发 Channel 成功注册事件； 接下来，如果 Channel 处于活动状态，且 Channel 是第一次注册，调用pipeline.fireChannelActive()触发 Channel 处于活动状态事件，同一个 Channel 只有在第一次注册的情况下才可能触发该事件；这样是为了防止多次触发 Channel Active 事件：比如可能存 在 Channel 注册后，调用 deregister，再次调用 register；如果 Channel 处于活动状态，且设置了 autoRead，则进行 beginRead；而这个 beginRead 方法里

@Override
    public final void beginRead() {
        assertEventLoop();
    
        if (!isActive()) {
            return;
        }
    
        try {
            doBeginRead();
        } catch (final Exception e) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireExceptionCaught(e);
                }
            });
            close(voidPromise());
        }
    }

关键方法就是 doBeginRead，这个方法主要实现在 AbstractNioChannel 类中，前面已经分析过这个方法。

***2、方法***  ***bind***

bind  方法主要用于绑定指定端口。对于服务端，用于绑定监听端口，并设置  backlog  参数；对于客户端，用于指定客户端 Channel  的本地绑定  Socket  地址。

@Override
    public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
        assertEventLoop();
    
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
    
        // See: https://github.com/netty/netty/issues/576
        if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
            localAddress instanceof InetSocketAddress &&
            !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
            !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
            // Warn a user about the fact that a non-root user can't receive a
            // broadcast packet on *nix if the socket is bound on non-wildcard address.
            logger.warn(
                    "A non-root user can't receive a broadcast packet if the socket " +
                    "is not bound to a wildcard address; binding to a non-wildcard " +
                    "address (" + localAddress + ") anyway as requested.");
        }
    
        boolean wasActive = isActive();
        try {
            doBind(localAddress);
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
    
        if (!wasActive && isActive()) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireChannelActive();
                }
            });
        }
    
        safeSetSuccess(promise);
    }

首先获取下通道的激活状态放入 wasActive，后面会用到，接下来执行具体的 doBind 方法，对于 NioSocketChannel 和 NioServerSocketChannel 有不同的实现，

@Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        doBind0(localAddress);
    }
    
    private void doBind0(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            SocketUtils.bind(javaChannel(), localAddress);
        } else {
            SocketUtils.bind(javaChannel().socket(), localAddress);
        }
    }

NioServerSocketChannel

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

如果绑定本地端口发生异常，则将异常设置到 ChannelPromise 中用于通知 ChannelFuture，随后调用 closeIfClosed 方法来关闭 Channel。 接下来如果是在绑定阶段成为 active 状态，则将调用 fireChannelActive 方法放进NioEventLoop 执行队列中。

***3、disconnect***  ***方法***

@Override
    public final void disconnect(final ChannelPromise promise) {
        assertEventLoop();
    
        if (!promise.setUncancellable()) {
            return;
        }
    
        boolean wasActive = isActive();
        try {
            doDisconnect();
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
    
        if (wasActive && !isActive()) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    pipeline.fireChannelInactive();
                }
            });
        }
    
        safeSetSuccess(promise);
        closeIfClosed(); // doDisconnect() might have closed the channel
    }

这个方法比较简单，关键的就是 doDisconnect();方法，这个方法对于 NioSocketChannel 和 NioServerSocketChannel 有不同的实现，

![20210215163423898.png][]

@Override
    protected void doDisconnect() throws Exception {
        throw new UnsupportedOperationException();
    }

NioSocketChannel 实现

![20210215163428173.png][]

@Override
    protected void doDisconnect() throws Exception {
        doClose();
    }
    
    @Override
    protected void doClose() throws Exception {
        super.doClose();
        javaChannel().close();
    }

很明显， NioServerSocketChannel  是无所谓断开连接一说的， NioSocketChanne  则调用了 SocketChannel 关闭连接。 然后调用 fireChannelInactive  方法。

***4、close***  ***方法***

@Override
    public final void close(final ChannelPromise promise) {
        assertEventLoop();
    
        close(promise, CLOSE_CLOSED_CHANNEL_EXCEPTION, CLOSE_CLOSED_CHANNEL_EXCEPTION, false);
    }

close 方法实际有点复杂，实际执行的是

private void close(final ChannelPromise promise, final Throwable cause,
                       final ClosedChannelException closeCause, final boolean notify) {
        if (!promise.setUncancellable()) {
            return;
        }
    
        if (closeInitiated) {
            if (closeFuture.isDone()) {
                // Closed already.
                safeSetSuccess(promise);
            } else if (!(promise instanceof VoidChannelPromise)) { // Only needed if no VoidChannelPromise.
                // This means close() was called before so we just register a listener and return
                closeFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        promise.setSuccess();
                    }
                });
            }
            return;
        }
    
        closeInitiated = true;
    
        final boolean wasActive = isActive();
        final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.
        Executor closeExecutor = prepareToClose();
        if (closeExecutor != null) {
            closeExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // Execute the close.
                        doClose0(promise);
                    } finally {
                        // Call invokeLater so closeAndDeregister is executed in the EventLoop again!
                        invokeLater(new Runnable() {
                            @Override
                            public void run() {
                                if (outboundBuffer != null) {
                                    // Fail all the queued messages
                                    outboundBuffer.failFlushed(cause, notify);
                                    outboundBuffer.close(closeCause);
                                }
                                fireChannelInactiveAndDeregister(wasActive);
                            }
                        });
                    }
                }
            });
        } else {
            try {
                // Close the channel and fail the queued messages in all cases.
                doClose0(promise);
            } finally {
                if (outboundBuffer != null) {
                    // Fail all the queued messages.
                    outboundBuffer.failFlushed(cause, notify);
                    outboundBuffer.close(closeCause);
                }
            }
            if (inFlush0) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        fireChannelInactiveAndDeregister(wasActive);
                    }
                });
            } else {
                fireChannelInactiveAndDeregister(wasActive);
            }
        }
    }
    
    private void doClose0(ChannelPromise promise) {
        try {
            doClose();
            closeFuture.setClosed();
            safeSetSuccess(promise);
        } catch (Throwable t) {
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

5、这个方法中主要实现了一下几个功能:

1) 、确保在多线程环境下，多次调用  close  和一次调用的影响一致，并且可以通过  promis 得到同样的结果。

2) 、保证在执行  close  的过程中，不能向  channel  写数据。

3) 、调用  doClose0  执行执真正的  close  操作。

4) 、调用  deregister  对  channel  做最后的清理工作，并触发  channelInactive, channelUnregistered 事件。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

主要是设置  close  动作的回调结果，

![20210215163529464.png][]

把发送缓存区  outboundBuffer  设置为  null,  这样，在  close  的过程中，所有  channel  的 write 方法都返回错误。

![20210215163539765.png][]

prepareToClose  默认实现是返回  null,  它是一个  protected  方法，可以根据需要覆盖它， 用来在关闭之前做一些准备工作，同时指定一个 executor ，让接下来的关闭动作都在这个 executor 中执行。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

接下来的  if..else  里其实实现的都是功能几乎一样的，只不过， if  语句中的将在prepareToClose 提供的  executor  中执行。调用  doClose0  执行关闭操作，清理  outboundBuffer,

调用  fireChannelInactiveAndDeregister  触发  channelInactive  和  channelDeregister  事件。而  else语句中，多了一点，通过 inFlush0  属性检查当前是否正在进程  flush  操作，如果是，使用 invokerLater 确保在  flush  操作完成之后再触发事件。doClose0 中是真正的关闭操作，它先调用  doClose ，然后设置  promise  的返回值。

***6、write***  ***方法***

write  方法实际上将消息添加到发送缓存区  ChannelOutboundBuffer  中，并不是真正的写 Channel。当调用  channel  的  write  方法写数据时，这个数据被一系列 ChannelOutboundHandler 处理之后，它被放进这个缓冲区中，并没有真正把数据写到 socket channel  中。然后再调用 channel 的  flush  方法， flush  会把  outboundBuffer  中数据真正写到  socket channel 。

@Override
    public final void write(Object msg, ChannelPromise promise) {
        assertEventLoop();
    
        ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null) {
            // If the outboundBuffer is null we know the channel was closed and so
            // need to fail the future right away. If it is not null the handling of the rest
            // will be done in flush0()
            // See https://github.com/netty/netty/issues/2362
            safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);
            // release message now to prevent resource-leak
            ReferenceCountUtil.release(msg);
            return;
        }
    
        int size;
        try {
            msg = filterOutboundMessage(msg);
            size = pipeline.estimatorHandle().size(msg);
            if (size < 0) {
                size = 0;
            }
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            ReferenceCountUtil.release(msg);
            return;
        }
    
        outboundBuffer.addMessage(msg, size, promise);
    }

首先是对发送缓存区 outboudBuffer 进行检查，如果是 null 抛出错误。调用 filterOutboundMessage 对 msg 进行过滤。这是一个 protected 方法，默认实现是什么都没做， 返回输入的 msg 参数。子类可以覆盖这个方法，把 msg 转换成期望的类型。然后计算 msg 的长度，再放入到 outboundBuffer 中。

***7、flush***  ***方法***

@Override
    public final void flush() {
        assertEventLoop();
    
        ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null) {
            return;
        }
    
        outboundBuffer.addFlush();
        flush0();
    }

flush 方法负责将发送缓冲区中待发送的消息全部写入到 Channel 中，并发送给通信对方。 flush 方法本身比较简单，首先将在发送缓存区 outboudBuffer 标识本次要发送消息的缓冲区

范围。然后调用  flush0  进行发送， flush0  中首先是一系列的检查，比如通道处于未激活，或者关闭状态，设置异常，关键方法是

doWrite(outboundBuffer) ; ，这个方法我们已经在前面的  NioSocketChannel  中分析过了。

**三、AbstractNioUnsafe**

AbstractNioUnsafe 是  AbstractUnsafe 类的 NIO 实现，它主要实现了  connect ,finishConnect 等方法。NioUnsafe  接口在  Unsafe  接口基础上增加了几个操作，包括访问  jdk  的 SelectableChannel、 socket  读数据等。

***1、connect***  ***方法  ***

@Override
    public final void connect(
            final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
    
        try {
            if (connectPromise != null) {
                // Already a connect in process.
                throw new ConnectionPendingException();
            }
    
            boolean wasActive = isActive();
            if (doConnect(remoteAddress, localAddress)) {
                fulfillConnectPromise(promise, wasActive);
            } else {
                connectPromise = promise;
                requestedRemoteAddress = remoteAddress;
    
                // Schedule connect timeout.
                int connectTimeoutMillis = config().getConnectTimeoutMillis();
                if (connectTimeoutMillis > 0) {
                    connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                            ConnectTimeoutException cause =
                                    new ConnectTimeoutException("connection timed out: " + remoteAddress);
                            if (connectPromise != null && connectPromise.tryFailure(cause)) {
                                close(voidPromise());
                            }
                        }
                    }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                }
    
                promise.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isCancelled()) {
                            if (connectTimeoutFuture != null) {
                                connectTimeoutFuture.cancel(false);
                            }
                            connectPromise = null;
                            close(voidPromise());
                        }
                    }
                });
            }
        } catch (Throwable t) {
            promise.tryFailure(annotateConnectException(t, remoteAddress));
            closeIfClosed();
        }
    }

首先获取当前的连接状态进行缓存，然后发起连接操作。整个方法中比较关键的就是doConnect(remoteAddress, localAddress) 方法

![20210215163702237.png][]

这个方法在前面的  NioSocketChannel  已经讲述过他的具体实现。

2、总结一下： 一般来说，SocketChannel  执行  connect() 操作有三种可能的结果。

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

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

(3 ）连接失败，直接抛出  IO  异常。

所以在  doConnect  方法中，如果是第 (2) 种结果，需要将  NioSocketChannel  中的 selectionKey 设置为  OP\_CONNECT ，监听连接应答消息。如果是第 (3) 种结果，就会关闭通道。

异步连接返回之后，需要判断连接结果。如果连接成功， doConnect  方法返回  true ，于是进入 fulfillConnectPromise(promise, wasActive); 方法，则触发  ChannelActive  事件，这个时

候发生的事情比较复杂，总的来说调用链如下：

pipeline().fireChannelActive();// 实际是  DefaultChannelPipeline

\--->HeadContext.channelActive

\--->HeadContext.readIfIsAutoRead

\--->channel.read();//  实际是  AbstractChannel

\--->pipeline().read(); // 实际是  DefaultChannelPipeline

\--->TailContext.read() // 没有覆盖，实际是  AbstractChannelHandlerContext

\--->HeadContext.read()//read  是个出站操作，最终由  HeadContext  执行

\--->unsafe.beginRead();// 实际是  AbstractUnsafe

\--->unsafe.doBeginRead // 实际是  AbstractNioUnsafe

所以，它最终会将  NioSocketChannel  中的  selectionKey  设置为  SelectionKey.OP\_READ ，用于监听网络读操作位。

如果没有立即连接上服务端，则执行分支。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

3、  上面的操作有两个目的。

(1 ）根据连接超时时间设置定时任务，超时时间到之后触发校验，如果发现连接并没有完成，则关闭连接句柄，释放资源，设置异常堆栈并发起去注册。

(2 ）设置连接结果监听器，如果接收到连接完成通知则判断连接是否被取消，如果被取消则关闭连接句柄，释放资源，发起取消注册操作。

***4、finishConnect***  ***方法***

@Override
    public final void finishConnect() {
        // Note this method is invoked by the event loop only if the connection attempt was
        // neither cancelled nor timed out.
    
        assert eventLoop().inEventLoop();
    
        try {
            boolean wasActive = isActive();
            doFinishConnect();
            fulfillConnectPromise(connectPromise, wasActive);
        } catch (Throwable t) {
            fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
        } finally {
            // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
            // See https://github.com/netty/netty/issues/1770
            if (connectTimeoutFuture != null) {
                connectTimeoutFuture.cancel(false);
            }
            connectPromise = null;
        }
    }

客户端接收到服务端的 TCP 握手应答消息，通过 SocketChannel 的 finishConnect 方法对连接结果进行判断。 首先缓存连接状态，当前返回 false，然后执行 doFinishConnect 方法判断连接结果。通过 SocketChannel 的 finishConnect 方法判断连接结果。

5、执行该方法返回三种可能结果。

1) 、连接成功返回  true;

2) 、连接失败返回  false;

3) 、发生链路被关闭、链路中断等异常，连接失败。

只要连接失败，就抛出  Error() ，由调用方执行句柄关闭等资源释放操作，如果返回成功，则执行 fulfillConnectPromise  方法，这个方法的作用前面  connect  方法中已经讲述过。 最后对连接超时进行判断: 如果连接超时时仍然没有接收到服务端的  ACK 应答消息﹐则由定时任务关闭客户端连接, 将  SocketChannel  从  Reactor  线程的多路复用器上摘除，释放资 源。

**四、NioMessageUnsafe**

这个是专门来处理客户端连接的  unsafe ，里面的方法很少，只有一个  read  方法。read 方法代码虽多，逻辑其实是比较简单的，首先是获取通道的配置、管道、接受缓冲区处理器等等，然后执行 NioServerSocketChannel  的  doReadMessages  方法，

![202102151638134.png][]

因为是专门来处理客户端连接的，所以参数的名字叫  readBuf ，但其实里面处理的都是一个个具体的客户端的连接。

**五、NioByteUnsafe**

NioByteUnsafe  负责具体的网络数据的读取，里面的方法不多， closeOnRead  处理读关闭 (半关闭 ) ， handleReadException  处理读异常，最关键的是  read  方法。

![20210215163837822.png][]

首先，获取  NioSocketChannel  的  SocketChannelConfig ，它主要用于设置客户端连接的TCP 参数。

@Override
        public final void read() {
            final ChannelConfig config = config();
            if (shouldBreakReadReady(config)) {
                clearReadPending();
                return;
            }
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
            allocHandle.reset(config);
    
            ByteBuf byteBuf = null;
            boolean close = false;
            try {
                do {
                    byteBuf = allocHandle.allocate(allocator);
                    allocHandle.lastBytesRead(doReadBytes(byteBuf));
                    if (allocHandle.lastBytesRead() <= 0) {
                        // nothing was read. release the buffer.
                        byteBuf.release();
                        byteBuf = null;
                        close = allocHandle.lastBytesRead() < 0;
                        if (close) {
                            // There is nothing left to read as we received an EOF.
                            readPending = false;
                        }
                        break;
                    }
    
                    allocHandle.incMessagesRead(1);
                    readPending = false;
                    pipeline.fireChannelRead(byteBuf);
                    byteBuf = null;
                } while (allocHandle.continueReading());
    
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
    
                if (close) {
                    closeOnRead(pipeline);
                }
            } catch (Throwable t) {
                handleReadException(pipeline, byteBuf, t, close, allocHandle);
            } finally {
                // Check if there is a readPending which was not processed yet.
                // This could be for two reasons:
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
                //
                // See https://github.com/netty/netty/issues/2254
                if (!readPending && !config.isAutoRead()) {
                    removeReadOp();
                }
            }
        }
    }

我们知道，既然是读取网络数据，肯定是需要缓冲区的，所以接下来是缓冲区的分配，然后读取通道内数据到缓冲区。 数据读取到缓冲区后，进行判断，如果没读到数据（lastBytesRead=0） 或者读取失败（返回值-1，表示发生了 I/O 异常，读取失败。），清空缓冲区或者中止继续读取，再触发ChannelRead 事件。

整个读取操作会放到一个循环中，进行多次读取，完成循环后，触发 ChannelReadComplete 事件

**六、NioSocketChannelUnsafe**

这个类的代码就很少了，里面只有一个  prepareToClose  方法，主要作用就是将事件取消注册（key.cancel() ）。

Unsafe 接口实现类相关源码分析完成，下篇我们分享   ，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/577e082044574a568d063e16ae43861c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/39c9e88bf4a7439aa1c6cd46b714a0c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/db75808719b74dc494f2311db295f6b9.png
[20210215163236897.png]: /images/20221024/4efac01f63ff43d9ada8e6b35fdede87.png
[20210215163423898.png]: /images/20221024/5cb9a9105178434d97bb37d2853b6ffc.png
[20210215163428173.png]: /images/20221024/ea354b4d8e784f00a7209fddfe433068.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/a1ed270d9eb7409296be7aa41b70110d.png
[20210215163529464.png]: /images/20221024/94e7e403eb4046c7a07949390f53f0f4.png
[20210215163539765.png]: /images/20221024/4f1293bb6f9148598d92d02794625091.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/c503ba5f6ef549038abee28e68cb20fa.png
[20210215163702237.png]: /images/20221024/ecc227119a7c44a29cb021c6616d56c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/3f43aba5b7e945b1ad2dc7bcfb7f3266.png
[202102151638134.png]: /images/20221024/19e18347c9c644b3be9bb8cfcd801233.png
[20210215163837822.png]: /images/20221024/4ece2fad59ef4dd78b3a91e1c6bc9e06.png