netty探索之旅四

上一篇我们研究了netty的客户端代码，这一篇我们研究一下服务端代码

以下是源码中服务端的启动代码
路径：example\src\main\java\io\netty\example\echo\EchoServer

public static void main(String[] args) throws Exception {
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }
        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)
             .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 LoggingHandler(LogLevel.INFO));
                     p.addLast(new EchoServerHandler());
                 }
             });

            ChannelFuture f = b.bind(PORT).sync();
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

和客户端的代码的层次比较相似。
NioEventLoopGroup：无论客户端还是服务端都需要指定。不过服务端中指定了两个NioEventLoopGroup，变量是bossGroup用于处理客户端的连接请求，变量workerGroup用于处理与各个客户端连接的IO操作。
Channel类型：因为是服务器端，所以是NioServerSocketChannel。

channel的初始化过程和客户端的大同小异。
同：结构基本一致
异：参数的类型不一样
客户端的Bootstrap和服务端的ServerBootstrap；客户端的NioSocketChannel和服务端的NioServerSocketChannel；

NioServerSocketChannel：

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

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

注意这里是openServerSocketChannel()。

this(newSocket(DEFAULT_SELECTOR_PROVIDER))：

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

这里调用父类构造器时,传入的参数是SelectionKey.OP_ACCEPT。客户端的NioSocketChannel传入的参数是SelectionKey.OP_READ。看过我关于Reactor和Proactor模式的文章中，就知道NIO是一种Reactor模式，通过selector来实现I/O的多路复用，服务端开始时需要监听客户端的连接请求，因此在这里我们设置了SelectionKey.OP_ACCEPT,即通知selector我们对客户端的连接请求事件感兴趣。

同样在AbstractChannel中会实例化一个unsafe和pipeline:值得注意的是客户端的unsafe是一个AbstractNioByteChannel：NioByteUnsafe的实例。服务端的NioServerSocketChannel是继承的AbstractNioMessageChannel类，在此类中重写了newUnsafe方法，

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

服务器端,unsafe是AbstractNioMessageChannel.AbstractNioUnsafe的实例。

小总结一下：
1，NioServerSocketChannel.newSocket(DEFAULT_SELECTOR_PROVIDER)打开一个Java NIO ServerSocketChannel。
2，AbstractChannel(Channel parent)中初始化AbstractChannel属性:
     parent 属性置为 null
     unsafe 通过newUnsafe()实例化一个unsafe对象,类型是 AbstractNioMessageChannel.AbstractNioUnsafe
     pipeline是new DefaultChannelPipeline(this)创建的实例.
3，AbstractNioChannel的属性:
     SelectableChannel ch 被设置为ServerSocketChannel.
     readInterestOp 被设置为 SelectionKey.OP_ACCEPT
     SelectableChannel ch 被配置为非阻塞的 ch.configureBlocking(false)
4，NioServerSocketChannel的属性:
     ServerSocketChannelConfig config = new NioServerSocketChannelConfig(this, javaChannel().socket())

ChannelPipeline初始化和channel的注册
这2块客户端和服务端过程是一样的。可以参考上一篇http://jishuaige.iteye.com/admin/blogs/2356798。

NioEventLoopGroup
在客户端的时候，我们只提供了一个NioEventLoopGroup对象，而在服务端我们初始化了2个NioEventLoopGroup对象，一个bossGroup，一个workerGroup。
bossGroup：用于服务端处理客户端的连接请求。类似于前台接待员。
workerGroup：负责客户端连接通道的IO操作。就是实际做事情的人。
bossGroup把人员领进门后，就可以等待下一个人员进来。人员进门之后workerGroup就负责对他进行服务了。

通过源码来看看这两位的工作：
EchoServer中：

 ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)

看看ServerBootstrap中的group方法

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;
    }

把bossGroup赋值给了ServerBootstrap的父类AbstractBootstrap的group(EventLoopGroup)变量。
把workerGroup赋值给了ServerBootstrap的childGroup(EventLoopGroup)变量。

 ChannelFuture f = b.bind(PORT).sync();

这个方法调用链是：
AbstractBootstrap.bind-->AbstractBootstrap.doBind--> AbstractBootstrap.initAndRegister
又看到了initAndRegister，前面在分析客户端的时候就看过了，里面有几段重要的代码

final Channel channel = channelFactory().newChannel();
    init(channel);
    ChannelFuture regFuture = group().register(channel);

newChannel()是一个NioServerSocketChannel实例，group()就是bossGroup(NioEventLoopGroup)，group().register(channel)【和客户端的一样】将NioServerSocketChannel和bossGroup关联起来(将ServerSocketChannel注册到eventLoop关联的selector上)。

接着我们来看看服务端是怎样接受客户端请求的。首先根据以上的分析，我们已经知道NioServerSocketChannel是对SelectionKey.OP_ACCEPT事件感兴趣。那么当ACCEPT事件发生的时候netty是在哪里进行处理的喃？
回到b.bind(PORT)的调用链AbstractBootstrap.bind -> AbstractBootstrap.doBind方法中可以看到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());
                }
            }
        });
    }

eventLoop()返回NioEventLoop对象，调用NioEventLoop.execute的方法

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

继续看里面的startThread()方法：

private void startThread() {
        if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                thread.start();
            }
        }
    }

thread对象是什么喃！还记得NioEventLoop的初始化方法不？在一系列的super方法后在SingleThreadEventExecutor的构造方法中

....................
thread = threadFactory.newThread(new Runnable() {
            @Override
            public void run() {
                boolean success = false;
                updateLastExecutionTime();
                try {
                    SingleThreadEventExecutor.this.run();
                    success = true;
.......................

对thread变量赋值一个匿名线程，回到startThread()方法中。这个方法中会将这个匿名线程启动起来。上面代码中run()方法会调用。注意：SingleThreadEventExecutor.this.run()
SingleThreadEventExecutor的子类NioEventLoop重写了run()方法，

@Override
    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:
                }

                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        processSelectedKeys();
                    } finally {
                        runAllTasks();
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally {
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }

1，轮询selector上的所有的channel的IO事件：select(wakenUp.getAndSet(false));
2，处理产生网络IO事件的channel：processSelectedKeys();
3，处理任务队列runAllTasks();
当发生了OP_ACCEPT事件就绪后，processSelectedKeys就开始处理就绪的事件，继续跟踪：
看到NioEventLoop中的processSelectedKey方法：

 private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            final EventLoop eventLoop;
            try {
                eventLoop = ch.eventLoop();
            } catch (Throwable ignored) {
                return;
            }
            if (eventLoop != this || eventLoop == null) {
                return;
            }
            unsafe.close(unsafe.voidPromise());
            return;
        }

        try {
            int readyOps = k.readyOps();NotYetConnectedException.
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }
            if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                ch.unsafe().forceFlush();
            }

            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }

 int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);
unsafe.finishConnect();

完成了客户端的连接操作，删除SelectionKey.OP_CONNECT事件。从以上代码看到OP_READ事件就绪后，会调用unsafe.read();就是调用到NioMessageUnsafe的read方法：看重点代码

int localRead = doReadMessages(readBuf);

继续调用到NioServerSocketChannel.doReadMessages方法：

@Override
    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = SocketUtils.accept(javaChannel());

        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;
    }

在doReadMessages中,通过javaChannel().accept()获取到客户端新连接的SocketChannel,接着就实例化一个NioSocketChannel,并且传入NioServerSocketChannel对象。这个方法后返回到NioMessageUnsafe的read方法中

 for (int i = 0; i < size; i ++) {
                    pipeline.fireChannelRead(readBuf.get(i));
                }

当有读取信息的后，通过ChannelPipeline机制,将读取事件逐级发送到各个handler中，这样就顺利完成数据读取。

上面的分析，我们已经知道了bossGroup和NioServerSocketChannel关联起来了，那么剩下的 workerGroup怎么关联喃？？？这个我们就要回到initAndRegister方法中接着往下看了：init方法。在ServerBootstrap中重写了init方法

@Override
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }

        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();

        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()));
        }

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

ch.eventLoop().execute方法会把这个匿名线程放到SingleThreadEventExecutor的任务队列中。请关注：pipeline.addLast(new ServerBootstrapAcceptor(currentChildGroup, currentChildHandler,currentChildOptions,currentChildAttrs));这句代码中的ServerBootstrapAcceptor对象(这句代码把ServerBootstrapAcceptor对象当成了一个handle放入了管道的最后)。childGroup对象就是前面的workerGroup赋值的值。

ServerBootstrapAcceptor中重写了channelRead方法

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

Channel child是一个NioSocketChannel，通过childGroup.register(child)这样workerGroup就和NioSocketChannel关联起来了。

workerGroup关联NioSocketChannel。
bossGroup关联NioServerSocketChannel。

看看ServerBootstrapAcceptor中的channelRead方法是怎么调用的。
上面提到了：当有读取信息的后，通过ChannelPipeline机制,将读取事件逐级发送到各个handler中的方法：

pipeline.fireChannelRead(readBuf.get(i));

这个代码会调用到DefaultChannelPipeline的fireChannelRead方法中：

@Override
    public final ChannelPipeline fireChannelRead(Object msg) {
        AbstractChannelHandlerContext.invokeChannelRead(head, msg);
        return this;
    }

继续：AbstractChannelHandlerContext的invokeChannelRead

static void invokeChannelRead(final AbstractChannelHandlerContext next, final Object msg) {
        ObjectUtil.checkNotNull(msg, "msg");
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRead(msg);
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRead(msg);
                }
            });
        }
    }

会调用next(HeadContext):AbstractChannelHandlerContext的方法invokeChannelRead，也是在AbstractChannelHandlerContext中：

private void invokeChannelRead(Object msg) {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRead(this, msg);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRead(msg);
        }
    }

handler()就会得到ServerBootstrapAcceptor对象，这样就调到了ServerBootstrapAcceptor重写的channelRead方法。

还记得我们上面的ServerBootstrap的init方法吗？在这个方法中添加了handler，方法里面通过handler()方法获取一个handler，不为空的情况下会添加到pipeline中，那么handler()得到的是什么对象喃？
其实handler()返回的就是我们在EchoServer类中添加的handler：

.handler(new LoggingHandler(LogLevel.INFO))

在ServerBootstrap初始化时的管道里面的handler情况是：
head---->ChannelInitializer--->tail

根据上一篇客户端的经验，当channel绑定到eventLoop后，这里是NioServerSocketChannel绑定到bossGroup中，会在pipeline中发出fireChannelRegistered事件,接着就会触发 ChannelInitializer.initChannel方法的调用这个方法被覆盖了的：

p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            ChannelHandler handler = handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }
            pipeline.addLast(new ServerBootstrapAcceptor(
                    currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
        }
    });

这个时候NioServerSocketChannel对应的管道里面就是：
head---->LoggingHandler--->ServerBootstrapAcceptor--->tail

在ServerBootstrapAcceptor.channelRead中会为新建的Channel设置handler并注册到一个 eventLoop中：

  final Channel child = (Channel) msg;
  child.pipeline().addLast(childHandler);
 childGroup.register(child).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            forceClose(child, future.cause());
                        }
                    }
                });

childHandler就是我们在启动的时候，设置的：

 .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());
                 }
             });

当这个客户端连接Channel注册到eventLoop后就会触发事件，调用ChannelInitializer.initChannel方法。那么新客户端连接后，NioSocketChannel对于的管道里面对应就是：
head---->sslHandler--->EchoServerHandler--->tail

看了服务端有2个handler：一个是通过handler()方法设置handler字段，一个是过 childHandler()设置childHandler字段。handler负责处理客户端的连接请求;childHandler 就是负责和客户端的连接的IO交互。

算是走完了服务端和客户端吧！但是还是比较混乱，只是跟着代码在走，还是跳跃的在走。有时自己都走晕了。至于netty为什么要这样写，这样写的优势，各种机制的实现还是没有分析到位，后面再努力吧！