美文网首页
Netty源码愫读(二)Channel相关源码学习

Netty源码愫读(二)Channel相关源码学习

作者: 桥头放牛娃 | 来源:发表于2018-11-20 22:02 被阅读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相关源码学习

          本文链接:https://www.haomeiwen.com/subject/lzffqqtx.html