Netty4之启动流程

灰太狼 2024-04-18 23:03 88阅读 0赞

Netty是基于JAVA NIO的网络应用框架，使用Netty可以迅速的开发网络应用。主要是用在服务端。这篇文章主要是分析Netty4.1.x的启动流程。通过启动流程可以更新清晰的知道Netty的运行逻辑。在介绍启动流程之前先说说几个名词的基本概念:

*  Bootstrap 启动器，负责对于整个Netty应用的启动，分为ServerBootstrap(Server端使用),Bootstrap(Client端使用);
 *  Channel是网络通信的主体，它负责同对端进行网络通信、注册和数据操作等功能包含Unsafe(实现对网络操作)、ChannelPipeline(channel的管道模型)、EventLoop(执行线程)等相关组件。简单来说对个Channel对应一个客户端的连接;
 *  ChannelPipeline(管道模型),用于存放Channel上下文信息。像处理器(Handler)、执行器(executor)以及Handler组成的双向链表AbstractChannelHandlerContext;
 *  ChannelHandler处理器，对Channel中上行或下行的数据进行逻辑处理，一般Handler分为InBound与OutBound两种。开发者也可以通过继承实现自已业务逻辑的Handler，如编解码、心跳等等。在用netty为框架进行开发时，这部分通常都需要开发者实现;
 *  AbstractChannelHandlerContext 处理器组成的双向链表，处理器的有序性就是通过这个来维护的;

上述几个是Netty很重要的组件，还有一些组件此处就不再叙述。这几个组建的关系是: 一个Channel对应一个ChannelPipeline,一个ChannelPipeline对应多个ChannelHandler,多个ChannelHandler会组成一个AbstractChannelHandlerContext放在ChannelPipeline中。注意一个ChannelHandler也可以对应多个ChannelPipeline。下面从一个简单的启动例子来分析启动过程:

public class NettyServer {
        public static void main(String[] args) throws Exception {
            new NettyServer().start("127.0.0.1", 8081);
        }
        public void start(String host, int port) throws Exception {
            ExecutorService executorService = Executors.newCachedThreadPool();
            EventLoopGroup bossGroup = new NioEventLoopGroup(0, executorService);//Boss I/O线程池,用于处理客户端连接,连接建立之后交给work I/O处理
            EventLoopGroup workerGroup = new NioEventLoopGroup(0, executorService);//Work I/O线程池
            EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(2);//业务线程池
            ServerBootstrap server = new ServerBootstrap();//启动类
            server.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, 3));
                    pipeline.addLast(businessGroup, new ServerHandler());
                }
            });
            server.childOption(ChannelOption.TCP_NODELAY, true);
            server.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
            server.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
            InetSocketAddress addr = new InetSocketAddress(host, port);
            server.bind(addr).sync().channel();//启动服务
        }
    }

例子中start()方法用于启动一个简单的NettyServer,server.bind()方法前面都是对ServerBootstrap启动器做一些初始变量的设置如I/O线程池、ChannelFactory及childHandler等相关属性的设置。下面看看server.bind()是如何处理的：

/**
     * Create a new {@link Channel} and bind it.
     */
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }
    
    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()方法来执行真正的绑定initAndRegister()是对ServerSocketChannel的初始化与注册。具体逻辑如下：

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
                channel.unsafe().closeForcibly();
                // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
                return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
            }
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
        }
    
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
    
        // If we are here and the promise is not failed, it's one of the following cases:
        // 1) If we attempted registration from the event loop, the registration has been completed at this point.
        //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
        // 2) If we attempted registration from the other thread, the registration request has been successfully
        //    added to the event loop's task queue for later execution.
        //    i.e. It's safe to attempt bind() or connect() now:
        //         because bind() or connect() will be executed *after* the scheduled registration task is executed
        //         because register(), bind(), and connect() are all bound to the same thread.
    
        return regFuture;
    }

首先用ChannelFactory新建了一个Channel,这个Channel的类型是通过Bootstrap.channel(NioServerSocketChannel.class)来指定的，然后执行init(channel)方法对这个新建的Channel进行初始化。具体代码如下，这个init()方法是ServerBootstrap或Bootstrap中定义的：

@Override
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }
    
        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;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
        }
    
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
    
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

首先为这个Channel设置option,这个option是通过Bootstrap.option()来设置的、然后设置attr,本文中的例子没有设置、再为这个Channel初始化一个ChannelPipeline。一般来说服务端的Channe会有ServerChannel跟Channel这两个，ServerChannel用于处理用户连接，连接建立好之后会交给Channel来处理这个连接后续的请求，ServerChannel会产生Channel，一个ServerChannel可以对应多个Channel。所以就需要的ServerChannel中存话Channel的一些设置项目。这个设置项通过 .childOption或.childHandler来实现的。设置完相关属性之后，就开始向Pipeline 添加ServerChannel的处理器。这里的initChannel是指新建NioServerSocketChannel，然后向这个Channel的pipeline中添加ServerBootstrapAcceptor处理器，重点来了，这个处理器用于建立对端的连接及初始化子Channel来处理这个连接中的后续请求。

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
    
        private final EventLoopGroup childGroup;
        private final ChannelHandler childHandler;
        private final Entry<ChannelOption<?>, Object>[] childOptions;
        private final Entry<AttributeKey<?>, Object>[] childAttrs;
        private final Runnable enableAutoReadTask;
    
        ServerBootstrapAcceptor(
                final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler,
                Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
            this.childGroup = childGroup;
            this.childHandler = childHandler;
            this.childOptions = childOptions;
            this.childAttrs = childAttrs;
    
            // Task which is scheduled to re-enable auto-read.
            // It's important to create this Runnable before we try to submit it as otherwise the URLClassLoader may
            // not be able to load the class because of the file limit it already reached.
            //
            // See https://github.com/netty/netty/issues/1328
            enableAutoReadTask = new Runnable() {
                @Override
                public void run() {
                    channel.config().setAutoRead(true);
                }
            };
        }
    
        @Override
        @SuppressWarnings("unchecked")
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            final Channel child = (Channel) msg;
    
            child.pipeline().addLast(childHandler);
    
            setChannelOptions(child, childOptions, logger);
    
            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }
    
            try {
                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);
            }
        }
    
        private static void forceClose(Channel child, Throwable t) {
            child.unsafe().closeForcibly();
            logger.warn("Failed to register an accepted channel: {}", child, t);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final ChannelConfig config = ctx.channel().config();
            if (config.isAutoRead()) {
                // stop accept new connections for 1 second to allow the channel to recover
                // See https://github.com/netty/netty/issues/1328
                config.setAutoRead(false);
                ctx.channel().eventLoop().schedule(enableAutoReadTask, 1, TimeUnit.SECONDS);
            }
            // still let the exceptionCaught event flow through the pipeline to give the user
            // a chance to do something with it
            ctx.fireExceptionCaught(cause);
        }
    }

ServerBootstrapAcceptor这个处理器ChannelRead方法中的msg其实就是一个子Channel，来ChannelRead方法被触发之后就初始化一个子Channel。final Channel child = (Channel) msg;,然后对这个Channel的相关属性进行设置。后续的请求就直接在这个Channel中进行了。这个方法被触发的条件是有服务端收到了客户端的请求。然后将childGroup注册到这个Channel上并为这个Channel分配一个EventLoop。init部分基本上是结束了，下面看看register部分.register部分主要是在AbstractChannel中定义的，具体如下:

@Override
    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);
            }
        }
    }
    
    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);
        }
    }

register部分首先也是对相关参数进行检查。eventLoop.inEventLoop()表示eventLoop为当前线程则直接执行真正的注册逻辑，否则新开一个线程去执行异步注册。register0中的doRegister();是真正执行注册的逻辑，主要通过Unsafe这个类实现对网络的注册产生SelectionKey，对网络编程有一定了解的人对这个一定很熟悉，这里就是讲了，有兴趣的可以去看看。执行完doRegister()这个方法后，pipeline.invokeHandlerAddedIfNeeded();这个方法会触发对于initChannel()这个方法的调用，从而产生Channel,后续的pipeline.fireChannelRegistered();就是向Channel中发送一些信息让相关Handler去执行。pipeline.invokeHandlerAddedIfNeeded()中调用

/**
     * {@inheritDoc} If override this method ensure you call super!
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isRegistered()) {
            // This should always be true with our current DefaultChannelPipeline implementation.
            // The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering
            // surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers
            // will be added in the expected order.
            if (initChannel(ctx)) {
    
                // We are done with init the Channel, removing the initializer now.
                removeState(ctx);
            }
        }
    }

来实现initChannel并将ChannelInitializer这个移除。init与register都执行完了，下面开始执行bind方法，bind方法的实现是在 AbstractUnsafe中实现的，具体如下：

@Override
    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);
    }

bind方法中会调用 doBind(localAddress)，这个方法是在NioServerSocketChannel中实现的

@Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

到这一步，Netty做的事情已经完成了， javaChannel().bind(localAddress, config.getBacklog()); 这个方法是调用java底层提供的方法进行真正的绑定。

总结：

1、ServerBootstrap主要是作为服务端的启动类，主要是产生ServerChannel用于处理客户端连接，产生Channel用于处理连接后续的请求。在初始化的时候可以通过.group为分别为这两种Channel指定不能的线程池，这样可以提高服务的吞吐量。

2、处理器Handler是双向链表存在于ChannelPipeline中，inBound跟outBound经过处理的顺序相反，Handler只对自己感兴趣的事件作处理。