2-netty源码分析之Client

2-netty源码分析之Client

其实记录netty客户端的启动过程基本跟server端相似,无非是借助Bootstrap组装组件,然后通过connect发起连接。前面的步骤不多记录,看server启动过程基本了解。

那么这里主要记录下connect相关细节,以及client发起connect后,server是如何处理的。
依然从demo出发:

EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.TCP_NODELAY, true)
     .handler(new ChannelInitializer<SocketChannel>() {
         @Override
         public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline p = ch.pipeline();
             if (sslCtx != null) {
                 p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
             }
             p.addLast(new EchoClientHandler());
         }
     });

    // Start the client.
    ChannelFuture f = b.connect(HOST, PORT).sync();

    // Wait until the connection is closed.
    f.channel().closeFuture().sync();
} finally {
    // Shut down the event loop to terminate all threads.
    group.shutdownGracefully();
}

可以看出前半部分基本一致,那就直接从下面这行核心代码出发:

ChannelFuture f = b.connect(HOST, PORT).sync();

1.connect分析
private ChannelFuture doConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    final ChannelPromise promise = channel.newPromise();
    if (regFuture.isDone()) {
        doConnect0(regFuture, channel, remoteAddress, localAddress, promise);
    } else {
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                doConnect0(regFuture, channel, remoteAddress, localAddress, promise);
            }
        });
    }

    return promise;
}

其实一眼看去做了这几件事:

  • 1.初始化并且注册channel,跟服务端类似,也就SocketChannel不一样等差别
  • 2.设置异步回调
  • 3.channel注册OK之后,执行doConnect操作,也就是doConnect0方法,这一步通过注册监听回调完成。

那么ChannelFuture何时出发operationComplete方法呢。从initAndRegister跟踪看看。


image.png

启动eventLoop线程执行register逻辑


image.png

跟着debug走到这里进去继续跟踪对于promise的相关设置


image.png

看到了promise.trySuccess()方法,继续走:


image.png
image.png

看到了notify方法体,


image.png

OK,再走一步 触发回调的逻辑就出现了:

private static void notifyListener0(Future future, GenericFutureListener l) {
    try {
        l.operationComplete(future);
    } catch (Throwable t) {
        logger.warn("An exception was thrown by " + l.getClass().getName() + ".operationComplete()", t);
    }
}

到这里紧接着就是执行开始设置的监听器里的doConnect逻辑了


image.png

,OK回调详细调用链很简单,一步步debug下去自然就很清楚。
无非是:

  • 1.将用户执行的线程转化为EventLoop线程
  • 2.执行channel register操作
  • 3.设置DefaultPromise属性,执行回调方法

2.doConnect
/** 发起连接就在这里啦 */
private static void doConnect0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.

    /** 注意:这里是eventLoop线程 */
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                if (localAddress == null) {
                    channel.connect(remoteAddress, promise);
                } else {
                    channel.connect(remoteAddress, localAddress, promise);
                }
                promise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

可以看到这里,一罐的做法,将真正执行发起连接的操作扔进任务队列,交由EventLoop去执行。此处没有设置localAdress,那么紧接着就进入到一下:


image.png

到这里,出现了比较熟悉的piepline,pipeline暂时不详细分析,后面一章会专门记录。那么这里就是交给管道去处理,其实可以猜想就是交给piepline中的handler处理:


image.png

果然如此,找到piepline链路的末节点tail处理执行,那么究竟这个connect具体是那个handler处理的呢?继续:

public ChannelFuture connect(final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    ...
    
    /**
     * 调用 findContextOutbound 方法, 从 DefaultChannelPipeline 内的双向链表的 tail 开始,
     * 不断向前寻找第一个 outbound 为 true 的 AbstractChannelHandlerContext, 然后调用它的 invokeConnect 方法
     *
     * 在 DefaultChannelPipeline 的构造器中, 会实例化两个对象: head 和 tail, 并形成了双向链表的头和尾.
     * head 是 HeadContext 的实例, 它实现了 ChannelOutboundHandler 接口, 并且它的 outbound 字段为 true.
     * 因此在 findContextOutbound 中, 找到的 AbstractChannelHandlerContext 对象其实就是 head.
     * 进而在 invokeConnect 方法中, 我们向上转换为 ChannelOutboundHandler --> invokeConnect 方法
     */
    final AbstractChannelHandlerContext next = findContextOutbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeConnect(remoteAddress, localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeConnect(remoteAddress, localAddress, promise);
            }
        }, promise, null);
    }
    return promise;
}

其实可以看到tail节点只不过是connect任务流转的第一个节点,这里会通过findContextOutbound去寻找第一个outbound为true的handler,然后由其去执行invokerConnect方法。还记的之前分析server时讲过:数据从外部流入对应的是inbound,相反则是outbound么.那么这里发起连接找得处理器必然是outbound对应的处理节点,SO 继续跟踪。


image.png

可以看到这里找到的是head节点,看下head类的描述:

final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler

果然继承了ChannelOutboundHandler,那么找到它就合理了。那么很自然最终执行connect就会进入HeadContext的connect方法了:

@Override
public void connect(
        ChannelHandlerContext ctx,
        SocketAddress remoteAddress, SocketAddress localAddress,
        ChannelPromise promise) throws Exception {

    /**  AbstractNioByteChannel.NioByteUnsafe */
    unsafe.connect(remoteAddress, localAddress, promise);
}

最终还不是具体的执行connect,交给了UnSafe去操作,其实这里也很合理,还记得之前分析Server时说过,netty最终与jdk nio底层打交道的所有事件,都是交给了UnSafe去操作的,那么这里就很合理这个设计了。
看看这里的UnSafe具体是哪个:


image.png

这里了是NioSocketChannelUnsafe,这里直接没有覆盖父类AbstractNioUnsafe的connect方法,那么下一步就直接进入了AbstractNioUnsafe的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();

        /** doConnect 子类实现 */
        if (doConnect(remoteAddress, localAddress)) {
            fulfillConnectPromise(promise, wasActive);
        } else {
            ...
        }
    } catch (Throwable t) {
        promise.tryFailure(annotateConnectException(t, remoteAddress));
        closeIfClosed();
    }
}
public boolean isActive() {
    SocketChannel ch = javaChannel();
    return ch.isOpen() && ch.isConnected();
}

我们这是刚启动clinet,那么这里的isActive根据代码就知道返回false,可以看到UnSafe调用了外部类AbstractNioChannel的doConnect方法,而这里doConnect的具体执行逻辑交由具体子类实现,很自然走到了NioSocketChannel(服务端对应的是NioServerSocketChannel).

@Override
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    if (localAddress != null) {
        doBind0(localAddress);
    }

    boolean success = false;
    try {
        boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            selectionKey().interestOps(SelectionKey.OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}

SocketUtils:

/**
 * 从 NioSocketChannel.newSocket 返回的 SocketChannel 对象;
 * 然后是调用 SocketChannel.connect 方法完成 Java NIO 层面上的 Socket 的连接.
 */
public static boolean connect(final SocketChannel socketChannel, final SocketAddress remoteAddress)
        throws IOException {
    try {
        return AccessController.doPrivileged(new PrivilegedExceptionAction<Boolean>() {
            @Override
            public Boolean run() throws IOException {
                return socketChannel.connect(remoteAddress);
            }
        });
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}

完成了jdk socket 的connect触发动作,连接发起也就算OK了,发起连接之后,如果此时连接尚未真正建立,设置interestOps为OP_CONNECT,这样就可以监听连接动作了,完成最终操作。

既然client已经发起了连接请求,那么接下来就要分析server如何处理连接请求了。


3.连接接入处理

其实这里的主要工作停留在server端,但是这个会直接决定整个client启动的成果,那么我们将server服务先跑起来,debug停留在reactor轮训那里,然后启动client发起请求。

那么服务端是怎么处理新连接的呢?主要如下:

  • 1.轮训出新连接
  • 2.处理新连接,其实这里会交给workGroup处理
  • 3.注册读事件

前面说把debug停留在轮训处,还记得服务端轮训的diamante段吗,直接里面有这样一个逻辑:

processSelectedKeys();
private void processSelectedKeys() {
    if (selectedKeys != null) {
        /** 处理优化过的selectedKeys */
        processSelectedKeysOptimized();
    } else {
        /** 正常的处理 */
        processSelectedKeysPlain(selector.selectedKeys());
    }
}
/**
 *  迭代 selectedKeys 获取就绪的 IO 事件, 然后为每个事件都调用 processSelectedKey 来处理它.
 *
 *  1.对于boss NioEventLoop来说,轮询到的是基本上就是连接事件,后续的事情就通过他的pipeline将连接扔给一个worker NioEventLoop处理
 *  2.对于worker NioEventLoop来说,轮询到的基本上都是io读写事件,后续的事情就是通过他的pipeline将读取到的字节流传递给每个channelHandler来处理
 *
 *
 *  netty的reactor线程第二步做的事情为处理IO事件,netty使用数组替换掉jdk原生的HashSet来保证IO事件的高效处理,
 *  每个SelectionKey上绑定了netty类AbstractChannel对象作为attachment,在处理每个SelectionKey的时候,
 *  就可以找到AbstractChannel,然后通过pipeline的方式将处理串行到ChannelHandler,回调到用户方法
 */
private void processSelectedKeysOptimized() {
    for (int i = 0; i < selectedKeys.size; ++i) {

        /** 1.取出IO事件以及对应的channel */
        final SelectionKey k = selectedKeys.keys[i];
        
        /** 取出后将数组置为空 */
        selectedKeys.keys[i] = null;

        final Object a = k.attachment();

        /** 处理该channel */
        if (a instanceof AbstractNioChannel) {
            processSelectedKey(k, (AbstractNioChannel) a);
        } else {
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task);
        }

        /**
         * 判断是否该再来次轮询
         * 也就是说,对于每个NioEventLoop而言,每隔256个channel从selector上移除的时候,就标记 needsToSelectAgain 为true,我们还是跳回到上面这段代码
         */
        if (needsToSelectAgain) {
            // See https://github.com/netty/netty/issues/2363
            /** 将selectedKeys的内部数组全部清空 */
            selectedKeys.reset(i + 1);

            /** 重新调用selectAgain重新填装一下 selectionKey */
            selectAgain();
            i = -1;
        }
    }
}
/**
 * processSelectedKey 中处理了三个事件, 分别是:

 * 1.OP_READ, 可读事件, 即 Channel 中收到了新数据可供上层读取.
 * 2.OP_WRITE, 可写事件, 即上层可以向 Channel 写入数据.
 * 3.OP_CONNECT, 连接建立事件, 即 TCP 连接已经建立, Channel 处于 active 状态.
 * 4.OP_ACCEPT,请求连接事件
 */
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final NioUnsafe unsafe = ch.unsafe();
    if (!k.isValid()) {
        final EventLoop eventLoop;
        try {
            eventLoop = ch.eventLoop();
        } catch (Throwable ignored) {
            return;
        }

        if (eventLoop != this || eventLoop == null) {
            return;
        }
        // close the channel if the key is not valid anymore
        unsafe.close(unsafe.voidPromise());
        return;
    }

    try {
        int readyOps = k.readyOps();
        // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
        // the NIO JDK channel implementation may throw a NotYetConnectedException.
        /**
         * 事件是 OP_CONNECT, 即 TCP 连接已建立事件.
         * 1.我们需要将 OP_CONNECT 从就绪事件集中清除, 不然会一直有 OP_CONNECT 事件.
         * 2.调用 unsafe.finishConnect() 通知上层连接已建立
         *  */
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
            // See https://github.com/netty/netty/issues/924
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            /**
             * unsafe.finishConnect() 调用最后会调用到 pipeline().
             * fireChannelActive(), 产生一个 inbound 事件, 通知 pipeline 中的各个 handler TCP 通道已建立(即 ChannelInboundHandler.channelActive 方法会被调用)
             */
            unsafe.finishConnect();
        }

        // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
            ch.unsafe().forceFlush();
        }

        // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
        // to a spin loop
        /** boos reactor处理新的连接   或者 worker reactor 处理 已存在的连接有数据可读 */
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {

            /** AbstractNioByteChannel中实现,重点 */
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

我们将上面各种事件处打上断点,启动client,看看第一次发起连接会以何种事件进入轮训处:


image.png

可以看到客户端第一次发情连接,会触发服务端Accept事件,紧接着会触发UnSafe的读处理,那么为什么是Accept,后面将解释;我们继续:
服务端对应的UnSafe是NioMessageUnsafe,前面已经跟踪过,那么这里自然会进入下面逻辑:

@Override
        public void read() {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            ...
            /** 拿到对应channel的pipeline */
            final ChannelPipeline pipeline = pipeline();
            boolean closed = false;
            Throwable exception = null;
            try {
                try {
                    for (;;) {

                        /** 读取一个连接,委托到外部类NioSocketChannel */
                        int localRead = doReadMessages(readBuf);
                        ...
                    }
                } catch (Throwable t) {
                    exception = t;
                }
                setReadPending(false);
                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    /** 每条新连接丢给服务端的channel */
                    pipeline.fireChannelRead(readBuf.get(i));
                }

                /** 清理资源 */
                readBuf.clear();
                pipeline.fireChannelReadComplete();

                if (exception != null) {
                    closed = closeOnReadError(exception);
                    pipeline.fireExceptionCaught(exception);
                }

                if (closed) {
                    if (isOpen()) {
                        close(voidPromise());
                    }
                }
            } finally {
                ...
            }
        }
    }

上面贴出比较核心的逻辑,看到这里:

int localRead = doReadMessages(readBuf);

@Override
protected int doReadMessages(List<Object> buf) throws Exception {

    /** 获取到客户端新连接的 SocketChannel , jdk底层操作,返回jdk底层nio创建的一条channel */
    SocketChannel ch = SocketUtils.accept(javaChannel());

    try {
        if (ch != null) {
            /** 将jdk的 SocketChannel 封装成自定义的 NioSocketChannel ,加到list */
            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;
}

读操作很简单,NioMessageUnsafe调用, 因为NioMessageUnsafe是与新连接相关, 因此就是调用jdk的accept()方法,新建立一条连接,同时将jdk的 SocketChannel 封装成自定义的 NioSocketChannel;同时跟踪到NioSocketChannel的构造器里会发现此时会对SelectionKey.OP_READ进行注册,即该channel监听事件是 SelectionKey.OP_READ

这段代码翻译一下:

  • 1.当一个 client 连接到 server 时, Java 底层的 NIO ServerSocketChannel 会有一个 SelectionKey.OP_ACCEPT 就绪事件, 接着就会调用到 這裡

  • 2.接着就实例化一个 NioSocketChannel, 并且传入 NioServerSocketChannel 对象(即 this), 由此可知, 我们创建的这个 NioSocketChannel 的父 Channel 就是 NioServerSocketChannel 实例 .

  • 3.接下来就经由 Netty 的 ChannelPipeline 机制, 将读取事件逐级发送到各个 handler 中, 于是就会触发前面我们提到的 ServerBootstrapAcceptor.channelRead 方法啦.

其实总结下就是doReadMessages 方法不断地读取连接,封装成NioSocketChannel放入 readBuf 容器,然后调用 pipeline.fireChannelRead(),将每条新连接丢给服务端的channel

OK,我们读取连接后,继续跟踪后面一个重要的逻辑

pipeline.fireChannelRead(readBuf.get(i));

image.png

可以看到,这里会调用AbstractChannelHandlerContext的invokeChannelRead方法。
这里对server启动做个回顾,方便介绍这里的调用链,我们在启动服务端的时候,piepiline里面的链路为:

head-->ServerBootstrapAcceptor-->tail链路,

p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) throws Exception {
        final ChannelPipeline pipeline = ch.pipeline();

        /** 这里的handler返回的是主.handler(new LoggingHandler(LogLevel.INFO)) 中的handler*/
        ChannelHandler handler = handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                /** 这里的 childGroup.register 就是将 workerGroup 中的某个 EventLoop 和 NioSocketChannel 关联 */
                pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});

就是上面这段代码。那么这里的invokeChannelRead会通过head往下流转,会进入ServerBootstrapAcceptor的channelRead方法,跟踪一下:


image.png

几经流转,确实到了这里:

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    /** 将msg转换成对应的channel */
    final Channel child = (Channel) msg;

    /** 添加用户自定义的childHandler */
    child.pipeline().addLast(childHandler);

    /** 设置 NioSocketChannel 对应的 attr和option */
    setChannelOptions(child, childOptions, logger);

    for (Entry<AttributeKey<?>, Object> e: childAttrs) {
        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
    }

    try {

        /** 将 workerGroup 中的某个 EventLoop 和 NioSocketChannel 关联 */
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

那这个逻辑做了什么呢?

  • 1.将msg转换成对应的channel
  • 2.添加用户自定义的childHandler
  • 3.将 workerGroup 中的某个 EventLoop 和 NioSocketChannel 关联

好像已经看到了workerGroup的身影,我们在server中强调过,server启动过程只包含bossGroup的启动,不涉及workerGroup的启动,那这里是不是就是workGroup的启动呢?
我们继续:
第二点说的childHandler就是server里的这段逻辑

.childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         if (sslCtx != null) {
             p.addLast(sslCtx.newHandler(ch.alloc()));
         }
         p.addLast(new EchoServerHandler());
     }
 });

记住,这里的channel是NioSocketChannel,不是我们启动服务端的那个NioServerSocketChannel了。
那么到这里,NioSocketChannel绑定的piepiline就是:

head-->ChannelInitializer-->tail

继续划重点:

childGroup.register(child)

很相似的一段代码,注册channel到group,不就是将一个指定的NioSocketChannel绑定到具体的EvenyLoop么,不也就是在WorkGroup线程池中选着一个去维护NioSocketChannel的相关事件么,好像是很明了,咱们继续看看register做了什么:

MultithreadEventLoopGroup#register方法:

@Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}
public EventExecutor next() {
    return chooser.next();
}

首先next方法就是调用执行器选择一个具体的EventLoop,紧接着用这个选择的EventLoop执行register操作。继续跟踪:

@Override
public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
    if (channel == null) {
        throw new NullPointerException("channel");
    }
    if (promise == null) {
        throw new NullPointerException("promise");
    }

    /** 获取 channel 的 unsafe() 底层操作对象, 然后调用它的 register. */
    channel.unsafe().register(this, promise);
    return promise;
}
image.png

可以看到这里的UnSafe是NioSocketChannelUnSafe类型,那么直接进入AbstractChannel#AbstractUnsafe的register方法:

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    ...
    /** 将eventLoop与channel绑定 */
    AbstractChannel.this.eventLoop = eventLoop;

    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            ...
        }
    }
}

嗯哼,核心点依然出现:

  • 1.将eventLoop与channel绑定
  • 2.通过eventLoop.execute方式启动线程
  • 3.以任务式完成注册
@Override
    protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            /** 将 Channel 对应的 Java NIO SockerChannel 注册到一个 eventLoop 的 Selector 中, 并且将当前 Channel 作为 attachment. */
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            ...
        }
    }
}

可以看到channel与selector进行了绑定,即selector对该channel进行轮训事件,并且所有该channel的事件都交由绑定的EventLoop执行。继续:

pipeline.invokeHandlerAddedIfNeeded();

这一步很熟悉哈,最终会触发啥可想而知:


image.png

此时才将用户自定义的handler装车进入piepline,那么这个时候的piepline会是这样:

head->EchoServerHandler->tail

继续


image.png

可以看到已经建立了连接,那么后面会干什么呢?看到下面就知道又是从head开始传递事件:

/** 为啥pipeline.fireChannelActive();最终会调用到AbstractNioChannel.doBeginRead(),了解pipeline中的事件传播机制 */
@Override
public final ChannelPipeline fireChannelActive() {
    /** 三次握手成功之后,pipeline.fireChannelActive();被调用,然后以head节点为参数,直接一个静态调用 */
    AbstractChannelHandlerContext.invokeChannelActive(head);
    return this;
}

active传递之后就会进入readIfIsAutoRead,又是一轮piepline传递,继续

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    ctx.fireChannelActive();

    readIfIsAutoRead();
}
image.png

进入到了AbstractChannel#AbstractUnsafe的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());
    }
}

可以看到注册了readInterestOp即SelectionKey.OP_READ事件,将 SelectionKey.OP_READ事件注册到selector中去,表示这条通道已经可以开始处理read事件了,也代表着客户端整个接入过程完毕。


4.总结
  • 1.客户端发起连接
  • 2.服务端reactor主线程轮训accept事件
  • 3.建立jdk底层的socketChannel 并封装成自己的NioSocketChannel,NioSocketChannel一旦建立,代表连接建立
  • 4.将NioSocketChannel绑定到指定的work EventLoop,并注册到指定的selector上
  • 5.注册读事件,开启处理
  • 6.完毕
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,445评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,889评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,047评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,760评论 1 276
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,745评论 5 367
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,638评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,011评论 3 398
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,669评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,923评论 1 299
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,655评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,740评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,406评论 4 320
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,995评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,961评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,197评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,023评论 2 350
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,483评论 2 342

推荐阅读更多精彩内容