netty详解-netty服务端启动流程分析

1.BootStrap和ServerBootStrap分别为客户端和服务端的启动类

分析netty服务端启动之前我们看一个使用java原生API进行NIO编程的例子：

public class NioServer3 {
    private ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    private Selector connectSelector = Selector.open();
    private Selector readSelector = Selector.open();
    public NioServer3() throws IOException {
    }

    public static void main(String[] args) throws IOException {
        NioServer3 server3 = new NioServer3();
        ConnectedTask connectedTask = new ConnectedTask(server3);
        ReadTask readTask = new ReadTask(server3);

        Thread connectThread = new Thread(connectedTask);
        connectThread.setName("connect-thread");
        Thread readThread = new Thread(readTask);
        readThread.setName("read-thread");

        connectThread.start();
        readThread.start();
    }
    
    public void processConnected(int port) throws IOException {
        serverSocketChannel.configureBlocking(false);
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(port));
        SelectionKey selectionKey1 = serverSocketChannel.register(connectSelector, 0);

        selectionKey1.interestOps(SelectionKey.OP_ACCEPT);
        while (true) {
            connectSelector.select();

            Set<SelectionKey> selectionKeys = connectSelector.selectedKeys();
            Iterator<SelectionKey> selectionKeyIterator = selectionKeys.iterator();
            while (selectionKeyIterator.hasNext()) {
                SelectionKey selectionKey = selectionKeyIterator.next();
                selectionKeyIterator.remove();
                if (selectionKey.isAcceptable()) {
                    ServerSocketChannel serverChannel = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel socketChannel = serverChannel.accept();
                    System.out.println(Thread.currentThread() + " new connected" + socketChannel.getRemoteAddress());
                    socketChannel.configureBlocking(false);
                    socketChannel.register(readSelector, SelectionKey.OP_READ);
                }
            }
        }
    }

    public void processRead() {
        ByteBuffer byteBuffer = ByteBuffer.allocate(512);
        while (true) {
            try {

                int selectNow = readSelector.selectNow();
                if (selectNow <= 0) {
                    continue;
                }
            } catch (IOException ioexception) {
                System.out.println("readSelector.selectNow");
                ioexception.printStackTrace();
                continue;
            }
            Set<SelectionKey> selectionKeys = readSelector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey next = iterator.next();

                if (next.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) next.channel();
                    if (socketChannel.isConnected()) {
                        try {
                            System.out.println(Thread.currentThread() + " new data from " + socketChannel.getLocalAddress());
                            byteBuffer.clear();
                            socketChannel.read(byteBuffer);
                            byteBuffer.flip();
                            Charset charset = Charset.forName("utf-8");
                            String receivedMessage =
                                    String.valueOf(charset.decode(byteBuffer));
                            System.out.println(socketChannel + ":" + receivedMessage);

                        } catch (IOException e) {
                            System.out.println("socketChannel.read(byteBuffer)");
                            try {
                                socketChannel.close();
                            } catch (IOException e1) {

                            }
                            e.printStackTrace();
                        }
                    }
                }

                iterator.remove();
            }

        }
    }

    public static class ConnectedTask implements Runnable {
        private NioServer3 server3;
        public ConnectedTask(NioServer3 server3) {
            this.server3 = server3;
        }
        @Override
        public void run() {
            try {
                server3.processConnected(8080);
            } catch (IOException e) {

            }
        }
    }

    public static class ReadTask implements Runnable {
        private NioServer3 server3;

        public ReadTask(NioServer3 server3) {
            this.server3 = server3;
        }
        @Override
        public void run() {
            server3.processRead();
        }
    }

上面的例子借鉴netty的思路，使用两个Selector，一个Selector处理新的连接，一个selector处理新连接建立之后的数据读写操作，同时开启两个线程处理不同的任务，connectThread处理新的连接，readThread处理客户端发送的数据，需要注意的是：

1. readThread里面使用readSelector.selectNow();而不是使用select方法，原因是如果同一个selector的register（这行代码socketChannel.register(readSelector, SelectionKey.OP_READ)）和select方法如果在不同的线程中会导致一个线程一直阻塞等待另一个线程的锁，而另一个线程由于没有新的事件到来而没有唤醒，同时所持有的锁一直释放不掉，可以将上面readTread里的selectNow改成select然后启动，并创建新的连接，查看线程堆栈信息看到该问题；
2. netty和这里不同的地方是新连接到来后的register方法是在worker线程调用的，这里为了简单起见没有这么处理；

BootStrap和ServerBootStrap为Netty编程中不可缺少的两个类所在的包如下：

bootstrappackage.png

BootStrap的类图如下:

BootStrapClass.png

可以看到BootStrap和ServerBootStrap都继承自AbstractBootStrap,BootStrap和ServerBootStrap大部分职责都相同，在AbstractBootStrap类中进行实现.Server端的代码一般如下：

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup)
                    .handler(new LoggingHandler(LogLevel.WARN))
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new MyServerInializer());

            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

一般会有几个固定的步骤（以NIO为例）：

1. 创建并制定bossGroup、workerGroup；
2. 指定handler（可以有也可以没有）；
3. 指定channel的类型（NIO一般都是NioServerSocketChannel.class），这个步骤会创建并指定的channelFactory，默认为ReflectiveChannelFactory，而channelFactory的newChannel返回channel的实例，ReflectiveChannelFactory的newChannel通过反射调用channel（channel方法指定的NioServerSocketChannel）的无参构造方法实例化channel，后面会讲到；
4. 指定childHandler，一般为ChannelInitializer的实现类，在ChannelInitializer的initChannel中指定一个或多个具体的channelHandler
5. 调用ServerBootStrap.bind方法

ServerBootStrap.bind为服务端启动最重要的一步，同时处理的逻辑也最为复杂，接下来重点分析ServerBootStrap.bind的实现，ServerBootStrap.bind调用的时序大致如下：

serverbind.png

io.netty.bootstrap.AbstractBootstrap#bind(java.net.SocketAddress)代码如下：

public ChannelFuture bind(SocketAddress localAddress) {
        //1.校验parentGoup和channelFactory是否为空
        //2.parentGroup在调用group方法的时候设置，channelFactory为调用channel方法是创建的ReflectiveChannelFactory实例
        //3.childHandler为空直接抛异常
        // childGroup为空则设置为parentGroup(即childGroup和parentGroup为同一个)
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }

其中validate方法主要做一些校验工作，主要的逻辑在doBind方法进行处理，接下来我们一步一步的分析dobind；
dobind的代码如下：

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

        if (regFuture.isDone()) {
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            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) {
                        promise.setFailure(cause);
                    } else {
                        promise.registered();

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

initAndRegister方法完成channel的初始化以及注册的过程，initAndRegister的代码如下：

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            //通过方法创建channel实例(为调用channel方法时传进去的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);
        }

        //往selector里注册channel
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }

1. channelFactory在调用channel时创建，具体类型为ReflectiveChannelFactory，channelFactory.newChannel()通过反射的调用channel的无参构造方法创建channel，channelFactory是可以自定义的，通过io.netty.bootstrap.AbstractBootstrap#channelFactory(io.netty.bootstrap.ChannelFactory<? extends C>)方法指定自定义的channelFactory
2. init方法主要完成以下几件事：
    1. 设置channelOption(如果有指定bootStrap的option)
    2. 设置attr（如果有指定bootstrap的attr）
    3. 往channel的pipeline中添加了一个自定义的ChannelInitializer，该ChannelInitializer的initChannel方法会在channel注册完成之后调用，而initChannel做了一件比较重要的事情就是往channel（NioServerSocketChannel）的pipeline里面添加了一个ServerBootstrapAcceptor，而ServerBootstrapAcceptor重写了channelRead方法（对于Server端来说，该方法在有新连接建立的时候调用），而这里面设置了新连接（对于ServerBootStrap可以说是childChannel）option、attr，同时调用了childGroup.register方法，在workerGroup中选出的eventLoop线程中进行注册流程，即本文开头的例子提到的新连接建立时的register
3. ChannelFuture regFuture = config().group().register(channel)，这句代码最终完成了channel的注册过程（此时并没有指定SelectionKey，SelectionKey最终在io.netty.channel.nio.AbstractNioChannel#doBeginRead方法指定）

接下来我们来看doBind0方法：

private static void doBind0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress localAddress, final ChannelPromise promise) {
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                if (regFuture.isSuccess()) {
                    channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                } else {
                    promise.setFailure(regFuture.cause());
                }
            }
        });
    }

该方法主要就是向事件循环中添加了一个任务，这个任务里面调用了bind方法，最终会调到io.netty.channel.socket.nio.NioServerSocketChannel#doBind方法，调用链路如下:

dobind的调用链路.png

doBind的代码如下：

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

可以看到最终是调用java的bind去了;

doBind方法之后执行了pipeline.fireChannelActive()，代码片段在io.netty.channel.AbstractChannel.AbstractUnsafe#bind里面

try {
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }

    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
        @Override
        public void run() {
            pipeline.fireChannelActive();
        }
    });
    

前面提到的指定SelectionKey（dobeginread）就是在这里，调用链路如下：

dobeginread的调用链路.png

到这里服务端启动（bind）的整个流程基本上完成；