Netty源碼閱讀（一） ServerBootstrap啟動

Netty源碼閱讀（一） ServerBootstrap啟動

轉自我的Github

Netty是由JBOSS提供的一個java開源框架。Netty提供異步的、事件驅動的網絡應用程序框架和工具，用以快速開發高性能、高可靠性的網絡服務器和客戶端程序。本文講會對Netty服務啟動的過程進行分析，主要關注啟動的調用過程，從這里面進一步理解Netty的線程模型，以及Reactor模式。

這是我畫的一個Netty啟動過程中使用到的主要的類的概要類圖，當然是用到的類比這個多得多，而且我也忽略了各個類的繼承關系，關於各個類的細節，可能以后會寫單獨的博客進行分析。在這里主要注意那么幾個地方：

1. ChannelPromise關聯了Channel和Executor，當然channel中也會有EventLoop的實例。
2. 每個channel有自己的pipeline實例。
3. 每個NioEventLoop中有自己的Executor實例和Selector實例。 

網絡請求在NioEventLoop中進行處理，當然accept事件也是如此，它會把接收到的channel注冊到一個EventLoop的selector中，以后這個channel的所有請求都由所注冊的EventLoop進行處理，這也是Netty用來處理競態關系的機制，即一個channel的所有請求都在一個線程中進行處理，也就不會存在跨線程的沖突，因為這些調用都線程隔離了。

下面我們先看一段Netty源碼里面帶的example代碼，直觀感受一下Netty的使用：

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100) // 設置tcp協議的請求等待隊列
             .handler(new LoggingHandler(LogLevel.INFO))
             .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());
                 }
             });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

 

首先我們先來了解Netty的主要類：

EventLoop 這個相當於一個處理線程，是Netty接收請求和處理IO請求的線程。

EventLoopGroup 可以理解為將多個EventLoop進行分組管理的一個類，是EventLoop的一個組。

ServerBootstrap 從命名上看就可以知道，這是一個對服務端做配置和啟動的類。

ChannelPipeline 這是Netty處理請求的責任鏈，這是一個ChannelHandler的鏈表，而ChannelHandler就是用來處理網絡請求的內容的。

ChannelHandler 用來處理網絡請求內容，有ChannelInboundHandler和ChannelOutboundHandler兩種，ChannlPipeline會從頭到尾順序調用ChannelInboundHandler處理網絡請求內容，從尾到頭調用ChannelOutboundHandler處理網絡請求內容。這也是Netty用來靈活處理網絡請求的機制之一，因為使用的時候可以用多個decoder和encoder進行組合，從而適應不同的網絡協議。而且這種類似分層的方式可以讓每一個Handler專注於處理自己的任務而不用管上下游，這也是pipeline機制的特點。這跟TCP/IP協議中的五層和七層的分層機制有異曲同工之妙。

現在看上面的代碼，首先創建了兩個EventLoopGroup對象，作為group設置到ServerBootstrap中，然后設置Handler和ChildHandler，最后調用bind()方法啟動服務。下面按照Bootstrap啟動順序來看代碼。

    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        super.group(parentGroup);
        if (childGroup == null) {
            throw new NullPointerException("childGroup");
        }
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = childGroup;
        return this;
    }

 

首先是設置EverLoopGroup，parentGroup一般用來接收accpt請求，childGroup用來處理各個連接的請求。不過根據開發的不同需求也可以用同一個group同時作為parentGroup和childGroup同時處理accpt請求和其他io請求。

    public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }

 

接下來的channel()方法設置了ServerBootstrap的ChannelFactory，這里傳入的參數是NioServerSocketChannel.class，也就是說這個ReflectiveChannelFactory創建的就是NioServerSocketChannel的實例。

后面的option()，handler()和childHandler()分別是設置Socket連接的參數，設置parentGroup的Handler，設置childGroup的Handler。childHandler()傳入的ChannelInitializer實現了一個initChannel方法，用於初始化Channel的pipeline，以處理請求內容。

之前都是在對ServerBootstrap做設置，接下來的ServerBootstrap.bind()才是啟動的重頭戲。我們繼續按照調用順序往下看。

    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }

      /**
     * Create a new {@link Channel} and bind it.
     */
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }

    // AbstractBootstrap
      private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // Registration future is almost always fulfilled already, but just in case it's not.
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    Throwable cause = future.cause();
                    if (cause != null) {
                        // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                        // IllegalStateException once we try to access the EventLoop of the Channel.
                        promise.setFailure(cause);
                    } else {
                        // Registration was successful, so set the correct executor to use.
                        // See https://github.com/netty/netty/issues/2586
                        promise.registered();

                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }

 

我們可以看到bind()的調用最終調用到了doBind(final SocketAddress)，在這里我們看到先調用了initAndRegister()方法進行初始化和register操作。了解JavaNIO框架的同學應該能看出來是在這個方法中將channel注冊到selector中的。最后程序再調用了doBind0()方法進行綁定，先按照順序看initAndRegister方法做了什么操作。

    // AbstractBootstrap
    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
          // ...
        }

        ChannelFuture regFuture = config().group().register(channel);
        // ...
        return regFuture;
    }

為了簡單其間，我忽略了處理異常分支的代碼，同學們有興趣可以自行下載Netty源碼對照。在這里終於看到channel的創建了，調用的是ServerBootstrap的channelFactory，之前的代碼我們也看到了這里的工廠是一個ReflectChannelFactory，在構造函數中傳入的是NioServerSocketChannel.class，所以這里創建的是一個NioServerSocketChannel的對象。接下來init(channel)對channel進行初始化。

    // ServerBootstrap
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            channel.config().setOptions(options);
        }
        
        // 設置channel.attr
        final Map<AttributeKey<?>, Object> attrs = attrs0();
        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();

        final EventLoopGroup currentChildGroup = childGroup;
        // childGroup的handler
        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()));
        }
        // 給channelpipeline添加handler
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                // group的handler
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }

                // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
                // In this case the initChannel(...) method will only be called after this method returns. Because
                // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
                // placed in front of the ServerBootstrapAcceptor.
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

先是設置了channel的option和attr，然后將handler加入到channelpipleline的handler鏈中，這里大家請特別注意ServerBootstrapAcceptor這個Handler，因為接下來對於客戶端請求的處理以及工作channl的注冊可全是這個Handler處理的。不過由於現在channel還沒有注冊，所以還不會調用initChannel()方法，而是將這個handler對應的context加入到一個任務隊列中，等到channel注冊成功了再執行。關於ChannelPipeline的內容我們以后再說。然后在initAndRegister()方法中調用config().group().register(channel)對channel進行注冊。config().group()獲取到的其實就是bossGroup，在這個例子中就是一個NioEventLoopGroup，由於它繼承了MultithreadEventLoopGroup所以這里調用的其實是這個類的方法。

    // MultithreadEventLoopGroup
    public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }

    public EventLoop next() {
        return (EventLoop) super.next();
    }

    // SingleThreadEventLoop
    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }

    @Override
    public ChannelFuture register(final ChannelPromise promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        promise.channel().unsafe().register(this, promise);
        return promise;
    }

這里會獲取EventLoopGroup中的一個EventLoop，其實我們用的是NioEventLoopGroup所以這里獲取到的其實是NioEventLoop，而NioEventLoop繼承了SingleThreadEventLoop，這里register方法調用的就是SingleThreadEventLoop中的方法。我們重遇來到了channel最終注冊的地方，這里其實是調用了channel的unsafe對象中的register方法，也就是NioServerSocketChannel的方法，這個方法是在AbstractChannel祖先類中實現的，代碼如下：

         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;
            }
            // 設置eventLoop
            AbstractChannel.this.eventLoop = eventLoop;
            // 這里是跟Netty的線程模型有關的，注冊的方法只能在channel的工作線程中執行
            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);
                }
            }
        }
      
      // AbstractNioChannel
      protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                return;
            } catch (CancelledKeyException e) {
              // ...
            }
        }
      }
    
    // AbstractSelectableChannel
    public final SelectionKey register(Selector sel, int ops,Object att)
        throws ClosedChannelException
    {
        synchronized (regLock) {
            if (!isOpen())
                throw new ClosedChannelException();
            if ((ops & ~validOps()) != 0)
                throw new IllegalArgumentException();
            if (blocking)
                throw new IllegalBlockingModeException();
            SelectionKey k = findKey(sel);
            if (k != null) {
                k.interestOps(ops);
                k.attach(att);
            }
            if (k == null) {
                // New registration
                synchronized (keyLock) {
                    if (!isOpen())
                        throw new ClosedChannelException();
                    k = ((AbstractSelector)sel).register(this, ops, att);
                    addKey(k);
                }
            }
            return k;
        }
    }

這里先設置了channel的eventLoop屬性，然后在接下來的一段代碼中判斷當前線程是否是channel的處理線程，也就是是不是eventLoop的線程，如果不是那么就將注冊作為一個任務用EventLoop.execute執行。按照這里的執行順序，當前線程肯定不是eventLoop的線程，所以會執行else分支，其實eventLoop的線程也是在這個調用中啟動的。最后的注冊是在AbstractSelectableChannel類的register()方法中執行的。這里有個很奇怪的地方，這里注冊的ops是0，也就是沒有感興趣的事件。這個地方我們后面在分析。

將channel注冊到selector的代碼就是這些了，我們回頭分析EventLoop.execute(…)，其實注冊的代碼是在這里面被調用的。

    // SingleThreadEventExecutor
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }

        boolean inEventLoop = inEventLoop();
        if (inEventLoop) {
            addTask(task);
        } else {
            startThread();
            addTask(task);
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }

如果當前線程是EventLoop的線程，就把task加到任務隊列中去，如果不是，那么啟動線程，然后再把task加入到任務隊列。

    // SingleThreadEventLoop
    private void startThread() {
        if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                doStartThread();
            }
        }
    }

    private void doStartThread() {
        assert thread == null;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }

                boolean success = false;
                updateLastExecutionTime();
                try {
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                    // Some clean work
                }
            }
        });
    }

其實最后的線程還是要落到EventLoop中的executor里面，而NioEventLoop初始化的時候executor屬性設置的是一個ThreadPerTaskExecutor，顧名思義也就是每個任務新建一個線程去執行，而在這個Task里面對EventLoop的thread屬性進行了設置，並且最后執行SingleThreadEventExecutor.this.run()，這個run方法在NioEventLoop中實現。

 protected void run() {
        for (;;) {
            try {
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;
                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));
                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                        // fallthrough
                }

                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    processSelectedKeys();
                    runAllTasks();
                } else {
                    final long ioStartTime = System.nanoTime();

                    processSelectedKeys();

                    final long ioTime = System.nanoTime() - ioStartTime;
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }

                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        break;
                    }
                }
            } catch (Throwable t) {
                logger.warn("Unexpected exception in the selector loop.", t);

                // Prevent possible consecutive immediate failures that lead to
                // excessive CPU consumption.
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // Ignore.
                }
            }
        }
    }
    
      private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
        if (selectedKeys.isEmpty()) {
            return;
        }

        Iterator<SelectionKey> i = selectedKeys.iterator();
        for (;;) {
            final SelectionKey k = i.next();
            final Object a = k.attachment();
            i.remove();

            if (a instanceof AbstractNioChannel) {
                // 處理ServerSocketChannl的事件，如accept
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }

            if (!i.hasNext()) {
                break;
            }

            if (needsToSelectAgain) {
                selectAgain();
                selectedKeys = selector.selectedKeys();

                // Create the iterator again to avoid ConcurrentModificationException
                if (selectedKeys.isEmpty()) {
                    break;
                } else {
                    i = selectedKeys.iterator();
                }
            }
        }
    }

這個就是Netty最后的Reactor模式的事件循環了，在這個循環中調用selector的select方法查詢需要處理的key，然后processSelectedKeys方法進行處理。在這里因為之前在注冊NioServerSocketChannel的時候把channel當作attachment當做attachment，所以如果key的attachement是AbstractNioChannel說明這個是ServerSocketChannel的事件，如connect，read，accept。

其實還有一些問題沒有寫清楚，如下：

1. ServerSocketChannel的interestOps的注冊
2. accept請求的處理
3. 線程模型
4. pipeline的鏈式調用
5. buffer 
。。。
這些我會繼續寫文章進行說明～(希望可以=_=)