Netty源码愫读（二）Channel相关源码学习

2018-11-20 本文已影响70人桥头放牛娃

io.netty.channel.Channel是Netty网络操作抽象类，它聚合了一组功能，包括但不限于网路的读，写，客户端发起连接，主动关闭连接，链路关闭，获取通信双方的网络地址等。它也包含了Netty框架相关的一些功能，包括获取该Channel的EventLoop，获取缓冲分配器ByteBufAllocator和pipeline等。

Unsafe是个内部接口，聚合在Channel中协助进行网络读写相关的操作，因为它的设计初衷就是Channel的内部辅助类，不应该被Netty框架的上层使用者调用，所以被命名为Unsafe。这里不能仅从字面理解认为它是不安全的操作，而要从这个架构的设计层面体会它的设计初衷和职责。

1、Channel功能说明

1.1、netty自定义Channel的原因

JDK的SocketChannel和ServerSocketChannel没有统一的Channel接口供业务开发者使用，对于用户而言，没有统一的操作视图，使用起来不方便。
JDK的SocketChannel和ServerSocketChannel的主要职责就是网络I/O操作，由于它们是SPI类接口，由具体的虚拟机厂家来提供，所以通过继承SPI功能类来扩展其功能的难度很大；直接实现ServerSocketChannel和SocketChannel抽象类，其工作量和重新开发一个新的Channel功能类是差不多的。
Netty的Channel需要能够跟Netty的整体架构融合在一起，例如I/O模型，基于ChannelPipeline的定制模型，以及基于元数据描述配置化的TCP参数等，这些JDK的SocketChannel和ServerSocketChanel都没有提供，需要重新封装。
自定义的Channel，功能实现更加灵活。

1.2、netty自定义Channel的设计理念

在Channel接口层，采用Facade模式进行统一封装，将网络I/O操作，网络I/O相关的其他操作封装起来，统一对外提供。
Channel接口的定义尽量大而全，为SocketChannel和ServerSocketChannel提供统一的视图，由不同子类实现不同的功能，公共功能在抽象父类中实现，最大程度地实现功能和接口的重用。
具体实现采用聚合而非包含的方式，将相关的功能类聚合在Channel中，由Channel统一分配和调度，功能实现更加灵活。

1.3、Channel主要API

接口名	描述
EventLoop eventLoop()	Channel需要注册到EventLoop的多路复用器上，用于处理I/O事件，通过eventLoop()方法可以获取到Channel注册的EventLoop。EventLoop本质上就是处理网络读写事件的Reactor线程。在Netty中，它不仅仅用来处理网络事件，也可以用来执行定时任务和用户自定义NioTask等任务。
ChannelPipeline pipeline()	返回channel分配的ChannelPipeline
boolean isActive()	判断channel是否激活。激活的意义取决于底层的传输类型。例如，一个Socket传输一旦连接到了远程节点便是活动的，而一个Datagram传输一旦被打开便是活动的
boolean isOpen()	判断Channel是否已经打开
boolean isRegistered()	判断Channel是否已经在对应的EventLoop中注册
SocketAddress localAddress()	返回本地的socket地址
SocketAddress remoteAddress()	返回远程的socket地址
Channel flush()	将之前已写的数据冲刷到底层Channel上去
boolean isWritable()	当且仅当I/O线程可以立即处理写请求时，返回true；当本方法返回false时，任何写操作将进行入队，直到i/o线程准备好处理队列中的写请求
ChannelMetadata metadata()	熟悉TCP协议的读者可能知道，当创建Socket的时候需要指定TCP参数，例如接收和发送的TCP缓冲区大小，TCP的超时时间。是否重用地址等。在Netty中，每个Channel对应一个物理链接，每个连接都有自己的TCP参数配置。所以，Channel会聚合一个ChannelMetadata用来对TCP参数提供元数据描述信息，通过metadata()方法就可以获取当前Channel的TCP参数配置。
Channel read()	从Channel中读取数据到第一个inBound缓冲区，当读取完毕，触发Handler的channelRead()事件，同时触发Handler的channelReadComplete()事件，以让Handler决定是否继续进行数据读取。如果有正在读取的操作，则此方法不做任何操作。
ChannelFuture closeFuture()	当Channel关闭时，通知对应的ChannelFuture。此方法总是返回同一个实例
Unsafe unsafe()	提供一个内部使用的类，此类实现了Unsafae相关接口
Channel parent()	对于服务端Channel而言，它的父Channel为空；对于客户端Channel，它的父Channel就是创建它的ServerSocketChannel。
ChannelId id()	返回ChannelId对象，ChannelId是Channel的唯一标识。
ChannelConfig config()	获取当前Channel的配置信息，例如CONNECT_TIMEOUT_MILLS。
long bytesBeforeUnwritable()	isWritable()返回false时，其返回可写的字节数；否则返回0
long bytesBeforeWritable()	isWritable()返回true时，其返回底层缓存未写的的字节数；否则返回0
ByteBufAllocator alloc()	返回内存分配器

2、Channel类继承图

Channel类继承图如下：

Channel类继承图.png

从类继承图可以看出：

（1）Channel是所有通用类的基础接口，ServerChannel是所有服务端Channel的通用接口；
（2）AbstractChannel为Channel的基础抽象类，其对Channel的一些通用功能做了简单实现；
（3）从AbstractChannel继承出基于不同协议及I/O类型的Channel实现类；

NioSocketChannel：异步I/O的客户端 TCP Socket 实现
NioServerSocketChannel：异步I/O的服务端 TCP Socket 实现
NioDatagramChannel：异步I/O的 UDP Socket 实现
NioSctpChannel：异步I/O的客户端 Sctp Socket 实现
NioSctpServerChannel：异步I/O的服务端 Sctp Socket 实现
EpollSocketChannel：基于linux的Epoll实现的事件驱动的客户端TCP Socket实现；
EpollServerSocketChannel：基于linux的Epoll实现事件驱动的服务端TCP Socket实现；
OioSocketChannel：同步I/O的客户端 TCP Socket 实现
OioServerSocketChannel：同步I/O的服务端 TCP Socket 实现
OioDatagramChannel：同步I/O的 UDP Socket 实现
OioSctpChannel：同步I/O的客户端 Sctp Socket 实现
OioSctpServerChannel：同步I/O的服务端 Sctp Socket 实现

3、AbstractChannel源码分析

AbstractChannel 是Channel的部分实现，维护了一个通道相关的资源，如channel id, pipeline等；而且实现了对该套接字的IO操作，以及设置interestOps；这里还没有牵扯到底层的细节，只是这个框架的结构。

3.1、成员变量

private static final ClosedChannelException FLUSH0_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractUnsafe.class, "flush0()");
    private static final ClosedChannelException ENSURE_OPEN_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractUnsafe.class, "ensureOpen(...)");
    private static final ClosedChannelException CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractUnsafe.class, "close(...)");
    private static final ClosedChannelException WRITE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractUnsafe.class, "write(...)");
    private static final NotYetConnectedException FLUSH0_NOT_YET_CONNECTED_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new NotYetConnectedException(), AbstractUnsafe.class, "flush0()");

    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;

    /** Cache for the string representation of this channel */
    private boolean strValActive;
    private String strVal;

FLUSH0_CLOSED_CHANNEL_EXCEPTION：当Channel已关闭时调用AbstractUnsafe的flush0()，设置此异常；
ENSURE_OPEN_CLOSED_CHANNEL_EXCEPTION ：当Channel已关闭时调用AbstractUnsafe的ensureOpen()，设置此异常；
CLOSE_CLOSED_CHANNEL_EXCEPTION ：当Channel已关闭时调用AbstractUnsafe的close()，设置此异常；
WRITE_CLOSED_CHANNEL_EXCEPTION：当Channel已关闭时调用AbstractUnsafe的write()，设置此异常；
FLUSH0_NOT_YET_CONNECTED_EXCEPTION ：当Channel未连接时调用AbstractUnsafe的flush0()，设置此异常；
parent：父Channel；
id：Channel对应的全局唯一ID；
unsafe：Unsafe实例；
pipeline：当前Channel对应的DefaultChannelPipeline；
unsafeVoidPromise ：异常通知，默认不使用。当我们不需要异常在pipeline中传播时，无需设置；
closeFuture：Channel关闭通知；
localAddress：本地地址；
remoteAddress：远程地址；
eventLoop：当前Channel注册的EventLoop；
registered：是否已经注册到EventLoop；
closeInitiated：关闭时的基础参数是否已设置；
strValActive：是否已经缓存Channel变成active之后的toString()值；
strVal：Channel的toString()值的缓存；

3.2、构造函数

Channel构造函数如下：

protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
}

id：通过newId()接口实现，默认实现方式为DefaultChannelId的newInstance()实现，DefaultChannelId的ID生成规则为：机器ID+处理器ID+序列号+时间戳+随机码；newId()为protected类型接口，可由子类重载；
unsafe：通过newUnsafe()接口实现，此方法为抽象方法，具体实现由子类实现；
pipeline：通过newChannelPipeline()接口实现，默认实现类为DefaultChannelPipeline，其设定了pipeline的头处理器（head）和尾处理器（tail）等基本信息；newChannelPipeline()接口未protected类型，可由子类重载；

3.3、核心方法

AbstractChannel的接口实现都比较简单，其具体实现基本都交由pipeline和unsafe进行处理。AbstractUnsafe是AbstractChannel的对应Unsafe接口实现。此处对AbstractUnsafe的核心接口实现进行分析。

3.3.1、register事件框架

register()实现源码：

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

此处对一些基本信息进行检查，如是否已经注册，EventLoop是否兼容等；同时判断当前线程是否与EventLoop在同一线程中，如果是则进行注册（register0()），否则将以任务方式将注册事件放入EventLoop的执行队列中，以防止多线程多线程并发情况。具体注册处理交由register0()。

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

注册流程如下：

设置不可取消表示，并确保Channel已经打开；
具体注册细节由子类的doRegister()实现；
确保用户的handler已经添加到pipeline中；
异步设置注册成功通知，并调用fireChannelRegistered()方法异步通知register事件；
对于服务端接受的客户端连接，如果首次注册，触发Channel的Active事件，如果已设置autoRead，则调用beginRead()开始读取数据。

beginRead()实现源码:

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

对EventLoop及Channel是否为active进行检查，调用doBeginRead()模板方法执行具体的处理；若处理异常，则异步调用fireExceptionCaught()方法，进行异常通知。

3.3.2、bind事件框架

bind()实现源码：

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

处理流程：

对基础的EventLoop、Channel是否打开等进行检查；
广播检测、地址检查、平台相关检查、权限检查等；
调用doBind()进行实际的绑定，具体有由子类实现；
如果是首次Active，则异步进行fireChannelActive()通知；

3.3.3、disconnect事件框架

disconnect()实现源码:

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
        }

处理流程：

EventLoop检查，设置不可取消等；
调用doDisconnect()进行实际的断开连接处理；
Channel从非Inactive变为Active时，异步调用fireChannelInactive()进行Inactive通知；

3.3.4、close事件框架

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

处理流程：

判断关闭的初始化信息是否设置，若未设置，当Channel已经关闭时，设置关闭成功通知；否则添加异步关闭完成监听器；
获取输出缓冲区ChannelOutboundBuffer及关闭执行线程；
若执行线程不为空，则在执行线程中添加任务，进行实际的关闭处理，并处理输出缓冲区的冲刷数据失败处理及fireChannelInactiveAndDeregister()事件处理；
若执行线程为空，则直接在本线程中执行关闭Channel的处理，并处理输出缓冲区的冲刷数据失败处理;
当inFlush0为true，即当前输出缓冲区正在冲刷数据，则异步进行fireChannelInactiveAndDeregister()事件通知，否则直接进行通知；

doColse0()实现源码：

private void doClose0(ChannelPromise promise) {
            try {
                doClose();
                closeFuture.setClosed();
                safeSetSuccess(promise);
            } catch (Throwable t) {
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

doClose()由具体子类实现。

3.3.5、deregister事件框架

deregister()实现源码：

private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {
            if (!promise.setUncancellable()) {
                return;
            }

            if (!registered) {
                safeSetSuccess(promise);
                return;
            }

            // As a user may call deregister() from within any method while doing processing in the ChannelPipeline,
            // we need to ensure we do the actual deregister operation later. This is needed as for example,
            // we may be in the ByteToMessageDecoder.callDecode(...) method and so still try to do processing in
            // the old EventLoop while the user already registered the Channel to a new EventLoop. Without delay,
            // the deregister operation this could lead to have a handler invoked by different EventLoop and so
            // threads.
            //
            // See:
            // https://github.com/netty/netty/issues/4435
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    try {
                        doDeregister();
                    } catch (Throwable t) {
                        logger.warn("Unexpected exception occurred while deregistering a channel.", t);
                    } finally {
                        if (fireChannelInactive) {
                            pipeline.fireChannelInactive();
                        }
                        // Some transports like local and AIO does not allow the deregistration of
                        // an open channel.  Their doDeregister() calls close(). Consequently,
                        // close() calls deregister() again - no need to fire channelUnregistered, so check
                        // if it was registered.
                        if (registered) {
                            registered = false;
                            pipeline.fireChannelUnregistered();
                        }
                        safeSetSuccess(promise);
                    }
                }
            });
        }

处理流程：

如果还未注册，则直接设置成功通知；否则异步进行注销处理；
异步调用doDeregister()进行注销处理，进行inactive事件通知处理，进行unregister的事件通知处理；

3.3.6、write事件框架

write()实现源码：

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

如果outboundBuffer为空，表示Channel正在关闭，则不进行写处理，直接设置写失败，并释放msg；filterOutboundMessage()为消息过滤器，由子类实现；outboundBuffer.addMessage()将消息添加到输出缓冲区中；

3.3.7、flush事件框架

flush()实现源码：

 public final void flush() {
            assertEventLoop();

            ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
            if (outboundBuffer == null) {
                return;
            }

            outboundBuffer.addFlush();
            flush0();
        }

对输出缓冲区器进行检查，为空则不进行处理；冲刷unflushedEntry中的数据；冲刷输出缓冲区的数据；

flush0()实现源码：

protected void flush0() {
            if (inFlush0) {
                // Avoid re-entrance
                return;
            }

            final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
            if (outboundBuffer == null || outboundBuffer.isEmpty()) {
                return;
            }

            inFlush0 = true;

            // Mark all pending write requests as failure if the channel is inactive.
            if (!isActive()) {
                try {
                    if (isOpen()) {
                        outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true);
                    } else {
                        // Do not trigger channelWritabilityChanged because the channel is closed already.
                        outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
                    }
                } finally {
                    inFlush0 = false;
                }
                return;
            }

            try {
                doWrite(outboundBuffer);
            } catch (Throwable t) {
                if (t instanceof IOException && config().isAutoClose()) {
                    /**
                     * Just call {@link #close(ChannelPromise, Throwable, boolean)} here which will take care of
                     * failing all flushed messages and also ensure the actual close of the underlying transport
                     * will happen before the promises are notified.
                     *
                     * This is needed as otherwise {@link #isActive()} , {@link #isOpen()} and {@link #isWritable()}
                     * may still return {@code true} even if the channel should be closed as result of the exception.
                     */
                    close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
                } else {
                    try {
                        shutdownOutput(voidPromise(), t);
                    } catch (Throwable t2) {
                        close(voidPromise(), t2, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
                    }
                }
            } finally {
                inFlush0 = false;
            }
        }

处理流程：

若正在冲刷数据，则不进行处理；
若输出缓冲区为空，则不进行处理；
设置inFlush0为true，表示正在冲刷数据；
若Channel为Inactive状态，则设置冲刷数据失败；
执行doWrite()进行实际的写数据，若写异常，设置相应的写失败；
最后设置inFlush0为false，表示完成数据冲刷；

4、AbstractNioChannel源码分析

AbstractNioChannel类继承图：

AbstractNioChannel类继承图.png

4.1、成员变量

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;

DO_CLOSE_CLOSED_CHANNEL_EXCEPTION：调用doClose()时错误的异常；
ch：NioSocketChannel和ServerSocketChannel的公共父类，用于设置参数和进行I/O操作；
readInterestOp：Read事件，服务端OP_ACCEPT，其他OP_READ
selectionKey：Channel注册到EventLoop后返回的选择键；
readPending：底层读事件标记
clearReadPendingRunnable：清除底层读事件标记任务
connectPromise：链接的异步结果
connectTimeoutFuture：连接超时检测任务异步结果
requestedRemoteAddress：连接的远端地址

4.2、核心方法

AbstractNioChannel的方法实现很简单，其主要是通过NioUnsafe接口进行实现的，而AbstractNioUnsafe为NioUnsafe的具体实现类，其接口继承图如下：

NioUnsafe类继承图.png

故只需要分析AbstractNioUnsafe的核心方法即可。

4.2.1、connect处理

connect()实现源码：

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

处理流程:

设置不可取消标志，确保Channel已经打开，检查连接异步监听是否已存在等；
调用doConnect()进行实际连接处理，此方法为抽象方法，由子类实现；
如果连接成功，则设置连接成功的异步通知；
如果设置连接超时时间，则添加超时任务，进行连接超时的异步通知；
添加异步连接通知的监听器，若异步结果被取消，则取消连接超时任务，清除连接的异步结果；

fulfileConnectPromise()实现源码：

private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
            // We still need to ensure we call fireChannelActive() in this case.
            boolean active = isActive();

            // trySuccess() will return false if a user cancelled the connection attempt.
            boolean promiseSet = promise.trySuccess();

            // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
            // because what happened is what happened.
            if (!wasActive && active) {
                pipeline().fireChannelActive();
            }

            // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
            if (!promiseSet) {
                close(voidPromise());
            }
        }
private void fulfillConnectPromise(ChannelPromise promise, Throwable cause) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // Use tryFailure() instead of setFailure() to avoid the race against cancel().
            promise.tryFailure(cause);
            closeIfClosed();
        }

fulfileConnectPromise()主要设置异步结果为成功，并出发Channel的Active事件。

finishConnect()实现源码：

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

finishConnect()只由EventLoop处理就绪selectionKey的OP_CONNECT事件时调用，从而完成连接操作。注意：连接操作被取消或者超时不会使该方法被调用。

4.2.2、doRegister处理

doRegister()实现源码：

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

    @Override
    protected void doDeregister() throws Exception {
        eventLoop().cancel(selectionKey());
    }

对于Register事件，将Channel注册到给定NioEventLoop的selector上即可。注意，其中第二个参数0表示注册时不关心任何事件，第三个参数为Netty的NioChannel对象本身。如果抛出CancelledKeyException 表示当前的Channel对应的SelectionKey已经被取消，此时立即对Selector进行select操作，其原因如下：

当select( )中的任意一种被调用时，如下步骤将被执行：

（1）已取消的键的集合将会被检查。如果它是非空的，每个已取消的键的集合中的键将从另外两个集合中移除，并且相关的通道将被注销。这个步骤结束后，已取消的键的集合将是空的。

（2）已注册的键的集合中的键的interest集合将被检查。在这个步骤中的检查执行过后，对interest集合的改动不会影响剩余的检查过程。

对于Deregister事件，选择键执行cancle()操作，选择键表示JDK Channel和selctor的关系，调用cancle()终结这种关系，从而实现从NioEventLoop中Deregister。需要注意的是：cancle操作调用后，注册关系不会立即生效，而会将cancle的key移入selector的一个取消键集合，当下次调用select相关方法或一个正在进行的select调用结束时，会从取消键集合中移除该选择键，此时注销才真正完成。一个Cancle的选择键为无效键，调用它相关的方法会抛出CancelledKeyException。

4.2.3、doRead处理

doRead()实现源码：

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

doRead()操作只是简单将read事件添加到SelectionKey中。

5、AbstractNioByteChannel源码分析

AbstractNioByteChannel类继承图如下：

[图片上传中...(NioByteUnsafe类继承图.png-376062-1542718590051-0)]

其对应的Unsafe类继承图如下：

NioByteUnsafe类继承图.png

AbstractNioByteChannel的实现很简单，本处主要分析其对应的NioByteUnsafe类。其是 NioSocketChannel 的父类，只有一个成员变量 flushTask，负责写半包消息。

5.1、核心方法

5.1.1、read处理

read()实现源码：

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

处理流程：

检查输入流是否已关闭或远端已经关闭，是则清除读处理；
调用doReadBytes()抽象方法读取数据；
若数据长度小于等于0，表示无数据可读，释放缓冲区；若读取的数据长度小于0表读出错，设置关闭相关标志；
数据读取完成后，pipeline进行read事件通知；
若需要关闭Channel，则调用closeOnRead()进行关闭处理；
若读处理异常，则调用handleReadException()进行读异常处理；

closeOnRead()实现源码：

private void closeOnRead(ChannelPipeline pipeline) {
            if (!isInputShutdown0()) {
                if (isAllowHalfClosure(config())) {
                    shutdownInput();
                    pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
                } else {
                    close(voidPromise());
                }
            } else {
                inputClosedSeenErrorOnRead = true;
                pipeline.fireUserEventTriggered(ChannelInputShutdownReadComplete.INSTANCE);
            }
        }

当Input流已未关闭，且Channel参数ALLOW_HALF_CLOSURE为True时，会触发用户事件ChannelInputShutdownEvent，否则，直接关闭该Channel；否则会触发用户事件ChannelInputShutdownReadComplete。

handleReadException()实现源码：

private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
                RecvByteBufAllocator.Handle allocHandle) {
            if (byteBuf != null) {
                if (byteBuf.isReadable()) {
                    readPending = false;
                    pipeline.fireChannelRead(byteBuf);
                } else {
                    byteBuf.release();
                }
            }
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();
            pipeline.fireExceptionCaught(cause);
            if (close || cause instanceof IOException) {
                closeOnRead(pipeline);
            }
        }

当缓存区有数据时，会触发read事件，触发读完成时间，触发异常时间；

5.1.2、write处理

doWrite()实现源码：

protected void doWrite(ChannelOutboundBuffer in) throws Exception {
  int writeSpinCount = config().getWriteSpinCount();
  do {
      Object msg = in.current();
      //无数据需要写，则清除SelectionKey的写标志
      if (msg == null) {
        // Wrote all messages.
        clearOpWrite();
        // Directly return here so incompleteWrite(...) is not called.
         return;
    }
      //调用doWriteInternal做实际的写处理
      writeSpinCount -= doWriteInternal(in, msg);
  } while (writeSpinCount > 0);
      incompleteWrite(writeSpinCount < 0);
}

此处处理比较简单，如果无数据可写，则清除SelectionKeyd的写事件；否则调用doWriteInternal()进行数据的写入；之后调用incompleteWrite()对是否写完成进行处理。

doWriteInternal()实现源码：

private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
  //msg类型为ByteBuf类型？
  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;
    }

  //msg类型为FileRegion类型？表示写文件数据
  } 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和FileRegion；如果为ByteBuf，则最终调用doWriteBytes()进行数据的写入；如果为fileRegion则最终调用doWriteFileRegion()对文件进行写入。两个写入方法都为抽象方法，由子类实现。

incompleteWrite()实现源码：

protected final void incompleteWrite(boolean setOpWrite) {
  // Did not write completely.
  // setOpWrite 为 true表示数据未写完，设置 SelectionKey 写标志位
  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
  // 启动清除flushTask任务，继续写半包消息
    eventLoop().execute(flushTask);
  }
}

6、NioSocketChannel源码分析

NioSocketChannel实现源码:

NioSocketChannel类继承图.png

NioSocketChannel为Netty的NIO底层客户端的具体实现类，一些具体的底层处理在此处实现。

6.1、核心方法

6.1.1、doBind操作

doBind()实现源码：

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

此处调用底层JDK进行Channel的绑定操作。

6.1.2、doConnect操作

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

    @Override
    protected void doFinishConnect() throws Exception {
        if (!javaChannel().finishConnect()) {
            throw new Error();
        }
    }

JDK中的Channel在非阻塞模式下调用connect()方法时，会立即返回结果：成功建立连接返回True，操作还在进行时返回False。返回False时，需要在底层OP_CONNECT事件就绪时，调用finishConnect()方法完成连接操作。

6.1.3、doWrite操作

protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        SocketChannel ch = javaChannel();
        int writeSpinCount = config().getWriteSpinCount();
        do {
            if (in.isEmpty()) {
                // 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.
            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:
                    // We have something else beside ByteBuffers to write so fallback to normal writes.
                    writeSpinCount -= doWrite0(in);
                    break;
                case 1: {
                    // 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: {
                    // 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);
    }

本段代码做的优化是：当输出缓冲区中有多个buffer时，采用Gathering Writes将数据从这些buffer写入到同一个channel。同时动态调整一次写的最大缓冲区大小。

7、AbstractNioMessageChannel源码分析

AbstractNioMessageChannel类继承图如下：

AbstractNioMessageChannel类继承图.png

7.1、核心方法

AbstractNioMessageChannel 是 NioServerSocketChannel、NioDatagramChannel 的父类。其主要方法也是 doWrite，功能和 AbstractNioByteChannel 的 doWrite 也类似，区别只是后者只处理 ByteBuf 和 FileRegion，前者无此限制，处理所有 Object。

7.1.1、doWrite

doWrite实现源码：

 protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        final SelectionKey key = selectionKey();
        final int interestOps = key.interestOps();

        for (;;) {
            Object msg = in.current();
            if (msg == null) {
                // Wrote all messages.
              //无数据可写则清除SelectionKey的写事件标志
                if ((interestOps & SelectionKey.OP_WRITE) != 0) {
                    key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
                }
                break;
            }
            try {
                boolean done = false;
                for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {
                    //doWriteMessage为抽象方法，由子类实现
                    if (doWriteMessage(msg, in)) {
                        done = true;
                        break;
                    }
                }

                if (done) {
                    in.remove();
                } else {
                    // Did not write all messages.
                    //未写完则设置SelectionKey的写标志位，等待底层可写时继续写入
                    if ((interestOps & SelectionKey.OP_WRITE) == 0) {
                        key.interestOps(interestOps | SelectionKey.OP_WRITE);
                    }
                    break;
                }
            } catch (Exception e) {
                if (continueOnWriteError()) {
                    in.remove(e);
                } else {
                    throw e;
                }
            }
        }
    }

doWriteMessage 方法在 NioServerSocketChannel 中实现如下所示，是因为 NioServerSocketChannel 只是用来监听端口，接收客户端请求，不负责传输实际数据。

NioServerSocketChannel中doWriteMessage()实现：

protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
    throw new UnsupportedOperationException();
}

7.1.2、read实现

read()实现源码：

public void read() {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            final ChannelPipeline pipeline = pipeline();
            final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
            allocHandle.reset(config);

            boolean closed = false;
            Throwable exception = null;
            try {
                try {
                    do {
                        //doReadMessages为抽象方法，由子类实现
                        int localRead = doReadMessages(readBuf);
                        //无数据可读？
                        if (localRead == 0) {
                            break;
                        }
                          //读取数据出错？
                        if (localRead < 0) {
                            closed = true;
                            break;
                        }
                        //增加消息数
                        allocHandle.incMessagesRead(localRead);
                    } while (allocHandle.continueReading());
                } catch (Throwable t) {
                    exception = t;
                }

                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    //触发pipeline的read事件，将读取到的数据交由应用层handler进行处理
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                readBuf.clear();
                allocHandle.readComplete();
                //触发pipeline的读取完成事件
                pipeline.fireChannelReadComplete();

                if (exception != null) {
                    closed = closeOnReadError(exception);
                    //触发pipeline的异常事件
                    pipeline.fireExceptionCaught(exception);
                }

                if (closed) {
                    inputShutdown = true;
                    if (isOpen()) {
                        close(voidPromise());
                    }
                }
            } 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();
                }
            }
        }
    }

8、NioServerSocketChannel源码分析

NioServerSocketChannel类继承图：

NioServerSocketChannel类继承图.png

NioServerSocketChannel 是服务端 Channel 的实现类，有一个用于配置 TCP 参数的 ServerSocketChannelConfig。

8.1、核心方法

NioServerSocketChannel为Netty的服务端NIO的实现，其只支持bind、read和close操作。

8.1.1、doBind实现

doBind()实现源码：

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

bind()的实现都是基于底层JDK实现的。

8.1.2、doReadMessage实现

doReadMessage()实现源码：

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

NioServerSocketChannel用于服务端的连接监听，此处的doReadMessages()方法每次最多返回一个消息（客户端连接），由此可知NioServerSocketChannel的read操作一次至多处理的连接数为config.getMaxMessagesPerRead()，也就是参数值MAX_MESSAGES_PER_READ。

8.1.3、doClose实现

doClose()实现源码：

protected void doClose() throws Exception {
    javaChannel().close();
}

调用底层JDK的Channel进行连接的关闭。

9、其他实现类

通过Channel的相关类继承图知道，Netty还有其他不同实现的Channel，除了NIO类型的，还有基于OIO的OioSocketChannel和OioServerSocketChannel，基于Epoll的EpollSocketChannel和EpollServerSocketChannel等，在此不多赘述。

相关阅读：
Netty源码愫读（一）ByteBuf相关源码学习【https://www.jianshu.com/p/016daa404957】
Netty源码愫读（三）ChannelPipeline、ChannelHandlerContext相关源码学习【https://www.jianshu.com/p/be82d0fcdbcc】
Netty源码愫读（四）ChannelHandler相关源码学习【https://www.jianshu.com/p/6ee0a3b9d73a】
Netty源码愫读（五）EventLoop与EventLoopGroup相关源码学习【https://www.jianshu.com/p/05096995d296】
Netty源码愫读（六）ServerBootstrap相关源码学习【https://www.jianshu.com/p/a71a9a0291f3】

参考书籍：

《Netty权威指南》第二版

参考博客：

https://www.jianshu.com/p/9258af254e1d
https://blog.csdn.net/vonzhoufz/article/details/39159193
https://hk.saowen.com/a/6cb46b200931fa05dbb222736de625a1d31e664e336a139f1f4794ff83d038ad
https://blog.csdn.net/lz710117239/article/details/77651209
https://juejin.im/post/5bda9ed6f265da3913474110

Netty源码愫读（二）Channel相关源码学习

1、Channel功能说明

1.1、netty自定义Channel的原因

1.2、netty自定义Channel的设计理念

1.3、Channel主要API

2、Channel类继承图

3、AbstractChannel源码分析

3.1、成员变量

3.2、构造函数

3.3、核心方法

3.3.1、register事件框架

3.3.2、bind事件框架

3.3.3、disconnect事件框架

3.3.4、close事件框架

3.3.5、deregister事件框架

3.3.6、write事件框架

3.3.7、flush事件框架

4、AbstractNioChannel源码分析

4.1、成员变量

4.2、核心方法

4.2.1、connect处理

4.2.2、doRegister处理

4.2.3、doRead处理

5、AbstractNioByteChannel源码分析

5.1、核心方法

5.1.1、read处理

5.1.2、write处理

6、NioSocketChannel源码分析

6.1、核心方法

6.1.1、doBind操作

6.1.2、doConnect操作

6.1.3、doWrite操作

7、AbstractNioMessageChannel源码分析

7.1、核心方法

7.1.1、doWrite

7.1.2、read实现

8、NioServerSocketChannel源码分析

8.1、核心方法

8.1.1、doBind实现

8.1.2、doReadMessage实现

8.1.3、doClose实现

9、其他实现类

猜你喜欢

热点阅读