zoukankan      html  css  js  c++  java
  • Netty4.0源码解析 NioServerSocketChannel

    一、引言
    Netty的Channel在JDK NIO的Channel基础上做了一层封装,提供了更多的功能。Netty的中的Channel实现类主要有:NioServerSocketChannel(用于服务端非阻塞地接收TCP连接)、NioSocketChannel(用于维持非阻塞的TCP连接)、NioDatagramChannel(用于非阻塞地处理UDP连接)、OioServerSocketChannel(用于服务端阻塞地接收TCP连接)、OioSocketChannel(用于阻塞地接收TCP连接)、OioDatagramChannel(用于阻塞地处理UDP连接):


    一个EventLoop一般持有多个Channel,每个EventLoop持有一个对应的线程,有这个线程负责处理这些Channel发出的事件。

    本篇文章我们先从服务端的NioServerSocketChannel开始分析:

    二、初始化过程
    NioServerSocketChannel在JDK的ServerSocketChannel的基础上做了一层封装。在NioServerSocketChannel的初始化过程中,可以简要地分为三步:
    1、实例化NioServerSocketChannel
    2、完成NioServerSocketChannel的管道初始化的第一步骤(向管道添加初始的ChannelHandler)
    3、将NioServerSocketChannel注册到NioEventLoopGroup,在此期间完成管道初始化的第二步骤(比如执行ChannelInitializer的handlerAdd方法)
    4、绑定端口,开始接受客户端连接。

    我们在ServerBootstrap进行引导的过程中,需要调用channel方法指定一个ServerChannel实现类的Class对象,channel方法随即会生成一个ChannelFactory工厂于ServerBootstrap实例中。NioServerSocketChannel的实例化在ServerBootstrap的bind方法中完成。bind方法调用的doBind方法中的initAndRegister方法中,会通过ChannelFactory实例构造出一个NioServerSocketChannel:

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
        	//通过ChannelFactory构造一个Channel实例
            channel = channelFactory().newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                channel.unsafe().closeForcibly();
                return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
            }
            return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
        }
    	//获取bossGroup实例并调用register方法绑定
        ChannelFuture regFuture = group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
        return regFuture;
    }
    

      通过ChannelFactory构造的Channel都是通过无参构造方法构造的,我们来分析NioServerSocketChannel的构造方法:

    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    

      这里通过静态newSocket方法构造了一个JDK的ServerSocketChannel:

    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException("Failed to open a server socket.", e);
        }
    }
    

      这里通过默认的SelectorProvider实例通过调用它的openServerSocketChannel构造了一个JDK的ServerSocketChannel实例,接着调用另外的重载的构造方法

    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
    

      

    NioServerSocketChannelConfig是NioServerSocketChannel的内部类,保存了以下内容:NioServerSocketChannel(this)、和刚刚构造的ServerSocketChannel通过socket方法获取的ServerSocket实例、引导过程通过options方法设置的参数。
    NioServerSocketChannel父类AbstractNioMessageChannel的构造方法:

    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent, ch, readInterestOp);
    }
    

      AbstractNioMessageChannel父类AbstractNioChannel构造方法:

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch; //传入JDK ServerSocketChannel
        this.readInterestOp = readInterestOp; //感兴趣通道事件为OP_ACCEPT
        try { //将ServerSocketChannel设为非阻塞
            ch.configureBlocking(false);
        } catch (IOException e) { //如果抛出异常那么关闭ServerSocketChannel
            try {
                ch.close();
            } catch (IOException e2) {
                if (logger.isWarnEnabled())
                    logger.warn("Failed to close a partially initialized socket.", e2);
            }
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }
    

      AbstractNioChannel父类AbstractChannel构造方法:

    protected AbstractChannel(Channel parent) {
        this.parent = parent; //ServerSocketChannel不存在父Channel
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }
    

      

    ServerSocketChannel不存在父Channel,所以parent为null,只有SocketChannel存在(其parent为ServerSocketChannel实例)。
    接着通过newUnsafe方法构造一个Channel.Unsafe接口实现类。在客户端引导过程已经提到过,Unsafe是Netty到JDK NIO的桥梁,Unsafe接口定义了以下方法:

    对于NioServerSocketChannel,其newUnsafe方法实现在父类AbstractNioMessageChannel中:

    @Override
    protected AbstractNioUnsafe newUnsafe() {
        return new NioMessageUnsafe();
    }
    

      

    NioMessageUnsafe继承了AbstractNioUnsafe,重写了read方法,用于处理OP_ACCPET事件。关于事件处理细节可以参考我的博客:https://blog.csdn.net/abc123lzf/article/details/83313530

    newChannelPipeline方法则是比较简单,直接构造一个DefaultChannelPipeline就完了。

    回到ServerBootstrap:当initAndRegister方法执行结束后,init方法随即会被调用并传入刚才构造的ServerSocketChannel实例作为参数:

    @Override
    void init(Channel channel) throws Exception {
    	//将通过option方法指定参数加入到Map中
        final Map<ChannelOption<?>, Object> options = options();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }
        //将初始化属性传入到Map中
        final Map<AttributeKey<?>, Object> attrs = attrs();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                channel.attr(key).set(e.getValue());
            }
        }
        ChannelPipeline p = channel.pipeline();
        //获取workGroup、childGroup指定的ChannelHandler
        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }
    	//向管道尾部添加一个ChannelInitializer实例
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = handler();
                //添加用户自定义的bossGroup的ChannelHandler
                if (handler != null)
                    pipeline.addLast(handler);
    			//向管道尾部添加一个ServerBootstrapAcceptor实例
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, 
                        		currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }
    

      

    init方法将引导过程通过options方法设置的参数导入到NioServerSocketChannel的ChannelConfig中,然后再将初始属性导入到NioServerSocketChannel间接父类DefaultAttributeMap中。

    完成上述步骤后,向ServerSocketChannel所属的管道尾部添加一个ChannelInitializer实例,它重写了initChannel方法,并指定向管道添加用户自定义的ChannelHandler(通过handler方法指定的),然后再异步添加一个ServerBootstrapAcceptor实例,用于将接收到的客户端连接对应的NioSocketChannel转交给workGroup进行管理。

    前几篇博客我们已经提到过,ChannelInitializer在完成自己的工作(执行完initChannel方法)后会将自己从管道中移除。但是就目前而言还仅仅只是完成了管道初始化的第一步骤,因为ChannelInitializer的handlerAdd方法还尚未调用。

    init方法执行完毕后,就会将这个ServerSocketChannel注册到bossGroup中,最终是通过AbstractUnsafe的register方法进行注册:

    @Override
    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        if (eventLoop == null)
            throw new NullPointerException("eventLoop");
        if (isRegistered()) { //不允许重复注册或者同一个Channel注册到不同的EventLoop中
            promise.setFailure(new IllegalStateException("registered to an event loop already"));
            return;
        }
        if (!isCompatible(eventLoop)) { //如果不能注册到这种类型的EventLoop中
            promise.setFailure(new IllegalStateException("incompatible event loop type: " +
            		eventLoop.getClass().getName()));
            return;
        }
        AbstractChannel.this.eventLoop = eventLoop;
        //如果当前线程就是eventLoop所属的线程,那么直接执行register0方法,否则异步执行
        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);
            }
        }
    }
    

      register方法在进行一系列参数检查和状态检查后,继而会执行register0方法:

    private void register0(ChannelPromise promise) {
        try {
            if (!promise.setUncancellable() || !ensureOpen(promise))
                return;
            boolean firstRegistration = neverRegistered; //默认为true
            doRegister(); //注册到Selector中
            neverRegistered = false;
            registered = true;
            pipeline.invokeHandlerAddedIfNeeded();
            safeSetSuccess(promise);
            //向管道发出Channel注册事件
            pipeline.fireChannelRegistered();
            if (isActive()) { //如果Channel是可用的
                if (firstRegistration) {
                	//向管道发出Channel可用事件
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    beginRead();
                }
            }
        } catch (Throwable t) {
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
    

      register0方法首先调用doRegister进行注册:

    @Override
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                if (!selected) {
                    eventLoop().selectNow();
                    selected = true;
                } else {
                    throw e;
                }
            }
        }
    }
    

      

    doRegister将ServerSocketChannel注册到了EventLoop的Selector上,并将附加对象设为this。doRegister方法执行结束后,就可以认为Channel已经注册到了EventLoop中,因为现在这个Channel持有的JDK Channel已经被EventLoop持有的Selector管理。
    随后,就会调用管道对象的invokeHandlerAddedIfNeeded方法:

    final void invokeHandlerAddedIfNeeded() {
        assert channel.eventLoop().inEventLoop();
        if (firstRegistration) {
            firstRegistration = false;
            callHandlerAddedForAllHandlers();
        }
    }
    

      

    invokeHandlerAddedIfNeeded此时就会调用callHandlerAddedForAllHandlers执行回调任务。

    让我们回顾下服务端的引导过程,当调用addLast方法将ChannelInitializer添加到管道中后,如果管道检测到这是它第一次调用addLast方法,并不会马上执行ChannelInitializer的handlerAdd方法去执行我们重写的方法,而是注册了一个回调任务,并把这个回调任务添加到了DefaultChannelPipeline的成员变量pendingHandlerCallbackHead中,如果有多个回调任务,那么它可以通过它的成员变量next去维护一个单向链表。

    了解这些后,我们再来看callHandlerAddedForAllHandlers这个方法的实现:

    private void callHandlerAddedForAllHandlers() {
        final PendingHandlerCallback pendingHandlerCallbackHead;
        synchronized (this) {
            assert !registered;
            registered = true; 
            pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
            this.pendingHandlerCallbackHead = null;
        }
        //遍历该链表,依次执行回调任务
        PendingHandlerCallback task = pendingHandlerCallbackHead;
        while (task != null) {
            task.execute();
            task = task.next;
        }
    }
    

      

    这样,我们指定的ChannelInitializer的handlerAdd方法随即会被调用,管道初始化的第二步骤也就随之完成。

    回到register0方法:
    接下来会依次向管道传递ChannelRegister、ChannelActive事件。传递完成后,NioServerSocketChannel开始进入端口绑定过程。
    回到ServerBootstrap的doBind方法,doBind方法会开始进行端口的绑定过程,最终会调用到NioServerSocketChannel的doBind方法:

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) { //如果JDK版本在1.7以上
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }
    

      

    这里直接调用到了JDK ServerSocketChannel的bind方法直接绑定指定的端口,并指定了最大连接数(默认为128),可以通过调用options并传入ChannelOption.SO_BACKLOG,然后指定一个最大连接数。当连接数超出后,客户端的连接请求会被阻塞。

    至此,NioServerSocketChannel初始化过程完成。

    三、事件处理
    初始化过程完成后,此时的NioServerSocketChannel已经被EventLoop的Selector所管理,Selector则由EventLoop所属的线程进行轮询。这个线程运行NioEventLoop的run方法,通过一个无限循环不停地处理IO事件和一般任务。

    当有客户端发起连接时,ServerSocketChannel会发出OP_ACCEPT事件,就会通过Unsafe的read方法(实现在AbstractNioMessageUnsafe)处理这个事件:

    @Override
    public void read() {
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        
        if (!config.isAutoRead() && !isReadPending()) {
            // ChannelConfig.setAutoRead(false) was called in the meantime
        	//从事件集中移除这个事件
            removeReadOp();
            return;
        }
        //获取每个循环读取消息的最大字节数
        final int maxMessagesPerRead = config.getMaxMessagesPerRead();
        final ChannelPipeline pipeline = pipeline();
        boolean closed = false;
        Throwable exception = null;
        try {
            try {
                for (;;) {
                	//将数据读到readBuf中,返回读取到的字节数
                    int localRead = doReadMessages(readBuf);
                    if (localRead == 0)
                        break;
                    if (localRead < 0) {
                        closed = true;
                        break;
                    }
                    if (!config.isAutoRead())
                        break;
                    //如果readBuf的长度大于maxMessagesPerRead,退出循环
                    if (readBuf.size() >= maxMessagesPerRead)
                        break;
                }
            } catch (Throwable t) {
                exception = t;
            }
            setReadPending(false);
            //获取读取到的ByteBuf数量,并通过循环将这些ByteBuf传递给ChannelInboundHandler
            int size = readBuf.size();
            for (int i = 0; i < size; i ++) {
                pipeline.fireChannelRead(readBuf.get(i));
            }
            //清除这些读取到的ByteBuf,并调用ChannelInboundHandler的fireChannelReadComplete方法
            readBuf.clear();
            pipeline.fireChannelReadComplete();
    
            //如果发生异常,调用ChannelInboundHandler的fireExceptionCaught方法
            if (exception != null) {
                closed = closeOnReadError(exception);
                pipeline.fireExceptionCaught(exception);
            }
    
            if (closed)
                if (isOpen())
                    close(voidPromise());
        } finally {
        	//再次检查这个事件有没有从事件集中去除
            if (!config.isAutoRead() && !isReadPending()) {
                removeReadOp();
            }
        }
    }
    

      

  • 相关阅读:
    Jmeter连接数据库并使用数据表数据作为接口所需参数
    使用CSV Data Set Config配置原件,参数化数据
    Jmeter连接SqlServer数据库并操作
    jmeter导入jmx文件报错:missing class com.thoughtworks.xstream.converters.ConversionException
    jmeter请求参数的两种方式
    badboy录制,出现弹框提示脚本错误解决方法
    (三)LoadRunner术语认识
    (二)LoadRunner目录分析
    7z命令
    python复制多层目录下的文件至其他盘符对应的目录中
  • 原文地址:https://www.cnblogs.com/jtlgb/p/10494731.html
Copyright © 2011-2022 走看看