欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

netty4源码分析-accept

程序员文章站 2022-04-22 18:04:43
...

转自 http://xw-z1985.iteye.com/blog/1941800

本文分析服务端如何accept客户端的connect请求,首先看下selector的I/O多路复用的分发逻辑:

//NioEventLoop  
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {  
        final NioUnsafe unsafe = ch.unsafe();  
        if (!k.isValid()) {  
            // close the channel if the key is not valid anymore  
            unsafe.close(unsafe.voidPromise());  
            return;  
        }  
        try {  
            int readyOps = k.readyOps();  
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {  
                unsafe.read();  
                if (!ch.isOpen()) {  
                    // Connection already closed - no need to handle write.  
                    return;  
                }  
            }  
            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();  
            }  
            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();  
            }  
        } catch (CancelledKeyException e) {  
            unsafe.close(unsafe.voidPromise());  
        }  
    }  

当有OP_ACCEPT事件到达时,分发给NioMessageUnsafe的read方法进行处理。

//NioMessageUnsafe  
public void read() {  
            assert eventLoop().inEventLoop();  
            final SelectionKey key = selectionKey();  
            if (!config().isAutoRead()) {  
                int interestOps = key.interestOps();  
                if ((interestOps & readInterestOp) != 0) {  
                    // only remove readInterestOp if needed  
                    key.interestOps(interestOps & ~readInterestOp);  
                }  
            }  
            final ChannelConfig config = config();  
            final int maxMessagesPerRead = config.getMaxMessagesPerRead();  
            final boolean autoRead = config.isAutoRead();  
            final ChannelPipeline pipeline = pipeline();  
            boolean closed = false;  
            Throwable exception = null;  
            try {  
                for (;;) {  
                    int localRead = doReadMessages(readBuf);  
                    if (localRead == 0) {  
                        break;  
                    }  
                    if (localRead < 0) {  
                        closed = true;  
                        break;  
                    }  
                    if (readBuf.size() >= maxMessagesPerRead | !autoRead) {  
                        break;  
                    }  
                }  
            } catch (Throwable t) {  
                exception = t;  
            }  
            int size = readBuf.size();  
            for (int i = 0; i < size; i ++) {  
                pipeline.fireChannelRead(readBuf.get(i));  
            }  
            readBuf.clear();  
            pipeline.fireChannelReadComplete();  
            if (exception != null) {  
                if (exception instanceof IOException) {  
                    // ServerChannel should not be closed even on IOException because it can often continue  
                    // accepting incoming connections. (e.g. too many open files)  
                    closed = !(AbstractNioMessageChannel.this instanceof ServerChannel);  
                }  
  
                pipeline.fireExceptionCaught(exception);  
            }  
  
            if (closed) {  
                if (isOpen()) {  
                    close(voidPromise());  
                }  
            }  
        }  
    }  

其中doReadMessages方法由NioServerSocketChannel实现:

// NioServerSocketChannel  
 protected int doReadMessages(List<Object> buf) throws Exception {  
        SocketChannel ch = javaChannel().accept();  
  
        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;  
    }  

SocketChannel ch = javaChannel().accept()就为接受的客户端连接建立了一个已连接套接字socketChannel.

buf.add(new NioSocketChannel(this, ch))会构造一个NioSocketChannel,并将其缓存到buf中(buf是一个List)。该NioSocketChannel的模式为非阻塞,readInterestOp为SelectionKey.OP_READ,并创建对应的管道和NioByteUnsafe实例。

maxMessagesPerRead表示如果此时有多个connect,那么只有当SeverSocketChannel建立的已连接套接字个数超过maxMessagesPerRead后,才会对每个已连接套接字触发channelRead事件。maxMessagesPerRead的默认值是16.

接下来分析channelRead事件做了什么事情:

channelRead是Inbound事件,会调用ServerBootstrapAcceptor的channelRead方法:

// ServerBootstrapAcceptor  
  public void channelRead(ChannelHandlerContext ctx, Object msg) {  
            Channel child = (Channel) msg;  
  
            child.pipeline().addLast(childHandler);  
  
            for (Entry<ChannelOption<?>, Object> e: childOptions) {  
                try {  
                    if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {  
                        logger.warn("Unknown channel option: " + e);  
                    }  
                } catch (Throwable t) {  
                    logger.warn("Failed to set a channel option: " + child, t);  
                }  
            }  
  
            for (Entry<AttributeKey<?>, Object> e: childAttrs) {  
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());  
            }  
  
            try {  
                childGroup.register(child);  
            } catch (Throwable t) {  
                child.unsafe().closeForcibly();  
                logger.warn("Failed to register an accepted channel: " + child, t);  
            }  
        }  

首先child.pipeline().addLast(childHandler)将服务端main函数中实例化的ChannelInitializer加入到管道中,该处理器的initChannel方法会在channelRegistered事件触发时被调用

childHandler(new ChannelInitializer<SocketChannel>() {  
                 @Override  
                 public void initChannel(SocketChannel ch) throws Exception {  
                     ch.pipeline().addLast(  
                             //new LoggingHandler(LogLevel.INFO),  
                             new EchoServerHandler());  
                 }  
             });  

然后设置NioSocketchannel的一些属性,最后进行注册:childGroup.register(child)。

这里采用的是childGroup,即worker线程池所在的Group,从Group中选择一个NioEventLoop,并启动其持有的worker线程,执行register0任务。

// AbstractUnsafe  
  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();  
                    promise.setFailure(t);  
                }  
            }  
        }  
  
        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 (!ensureOpen(promise)) {  
                    return;  
                }  
                doRegister();  
                registered = true;  
                promise.setSuccess();  
                pipeline.fireChannelRegistered();  
                if (isActive()) {  
                    pipeline.fireChannelActive();  
                }  
            } catch (Throwable t) {  
                // Close the channel directly to avoid FD leak.  
                closeForcibly();  
                closeFuture.setClosed();  
                if (!promise.tryFailure(t)) {  
                    logger.warn(  
                            "Tried to fail the registration promise, but it is complete already. " +  
                                    "Swallowing the cause of the registration failure:", t);  
                }  
            }  
        }  

此时worker线程就启动了。Register0任务在connect文章中已经描述,其主要功能就是将socketchannel注册到selector中;然后触发channelRegistered事件,调用ChannelInitializer的initChannel方法将服务端main函数中设置的处理器(本例为EchoServerHandler)加入到管道中,并将自己ChannelInitializer从管道中移除;最后触发channelActive事件,将ops设置为read。

// DefaultChannelPipeline  
  public ChannelPipeline fireChannelActive() {  
        head.fireChannelActive();  
  
        if (channel.config().isAutoRead()) {  
            channel.read();  
        }  
  
        return this;  
    }  

到此,worker线程对应的selector就开始监听该socketChannel上的read事件了。

接下来继续分析boss线程的执行:

将本次readBuf中缓存的所有NioSocketChannel注册后,就将他们从readBuf中移除。然后触发ChannelReadComplete事件,

// DefaultChannelPipeline  
 public ChannelPipeline fireChannelReadComplete() {  
        head.fireChannelReadComplete();  
        if (channel.config().isAutoRead()) {  
            read();  
        }  
        return this;  
    }  

head.fireChannelReadComplete()触发的是一个inbound事件,没有做任何事情。接着分析后续触发的read事件,这是一个outbound事件,也没有做任何事情(将ops重新设置为OP_ACCEPT,其实本来就是OP_ACCEPT)。

到此,一次accept的流程就执行完了。

总结:

一次accept的流程发生了以下事情:

  1. 为接受的客户端连接建立一个已连接套接字,设置为非阻塞。基于已连接套接字实例化一个NioSocketChannel,设置readInterestOp为SelectionKey.OP_READ,为其创建管道,并实例化内部的NioByteUnsafe。
  2. 在触发ServerSocketChannel的管道的channelRead方法之前,一个ServerSocketChannel一次可以最多缓存maxMessagesPerRead(默认为16)个NioSocketChannel。
  3. channelRead是一个Inbound事件,做了以下几件事:调用ServerBootstrapAcceptor处理器的channelRead方法为NioSocketChannel的管道加入ChannelInitializer处理器(该处理器的initChannel方法会在channalRegistered事件被触发时调用,将EchoServerHandler加入到管道中);设置NioSocketChannel的属性;从worker线程池中启动一个worker线程,执行register0任务。
  4. register0任务做的事情是:将socketChannal注册到selector中,触发channelRegistered事件,调用ChannelInitializer的initChannel方法将main函数中设置的处理器(譬如:EchoServerHandler)加入到管道中,然后触发channelActive事件,最后里面触发read事件,将ops设置为read。到此,worker线程所属的NioEventLoop持有的selector就开始监听socketChannel的read事件了。
  5. 最后触发ChannelReadComplete(inbound)事件,里面又会触发read(outbound)事件,这两个事件均没有做任何实事。
相关标签: netty4 accept