netty服务端的创建

服务端的创建

示例代码

netty源码中有一个netty-example项目，不妨以经典的EchoServer作为楔子。

// 步骤1

EventLoopGroup bossGroup = new NioEventLoopGroup(1);

EventLoopGroup workerGroup = new NioEventLoopGroup();

final EchoServerHandler serverHandler = new EchoServerHandler();

try {

    // 步骤2

    ServerBootstrap b = new ServerBootstrap();

    b.group(bossGroup, workerGroup)

        .channel(NioServerSocketChannel.class)

        .option(ChannelOption.SO_BACKLOG, 100)

            .attr(AttributeKey.newInstance("attr"), "attr")

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

            }

        })

        .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)

        .childAttr(AttributeKey.newInstance("childAttr"), "childAttr");

        // 步骤3

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

        // 步骤4

        f.channel().closeFuture().sync();

} finally {

    bossGroup.shutdownGracefully();

    workerGroup.shutdownGracefully();

}

整个流程可以分为几个步骤

创建workerGroup和bossGroup
创建ServerBootstrap，并设置参数
通过Serverbootstrap引导启动，并绑定端口。这里又可以分为初始化、注册、绑定端口、注册感兴趣事件4个小步骤
主线程阻塞
设置优雅关闭

步骤1留待下一节，先看步骤2

步骤2

ServerBootstrap首先设置bossGroup为parentGroup，workerGroup为childGroup，然后在channel方法中创建了一个工厂，该生产通过设置进来的泛型,利用反射来生产对象。此处生产对象为ServerSocketChannel。

除此之外还设置了handler、childHandler、option、childOption、attr、childAttr6个属性，根据group的命名规则，可以猜测不带child的属性是给bossGroup内的nioEventLoop使用，而以child开头的属性是给workGroupo内的nioEventLoop使用的。

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {

    super.group(parentGroup);

    this.childGroup = childGroup;

    return this;

}

public B channel(Class<? extends C> channelClass) {

    return channelFactory(new ReflectiveChannelFactory<C>

            (channelClass)

    ));

}

public ReflectiveChannelFactory(Class<? extends T> clazz) {

    this.constructor = clazz.getConstructor();

}

步骤3

serverBootstrap配置完后,开始服务端真正的启动工作，进入b.bind()方法，一路跳转到AbstractBootstrap.doBind(SocketAddress localAddress)方法。

private ChannelFuture doBind(final SocketAddress localAddress) {

    // 初始化并注册

    final ChannelFuture regFuture = initAndRegister();

    final Channel channel = regFuture.channel();

    //若已经注册完成,则开始绑定，否则添加一个监听器,待注册完成后绑定

    if (regFuture.isDone()) {

        ChannelPromise promise = channel.newPromise();

        doBind0(regFuture, channel, localAddress, promise);

        return promise;

    } else {

        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);

        regFuture.addListener((ChannelFutureListener) future ->

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

        return promise;

    }

}

步骤3.1

先看看initAndRegister()方法。首先是创建NioServerSocketChannel，通过工厂模式实例化一个channel。在channel的构造函数里，传入OP_ACCEPT感兴趣事件，并设置NioServerSocketChannel为非阻塞模式。在顶级抽象父类里，创建好id、unsafe、pipeline。unsafe对象是关联socket或其他进行IO操作组件的一个类，与jdk的unsafe对象类似，一般不需要用户关注

final ChannelFuture initAndRegister() {

    Channel channel  = channelFactory.newChannel();

    init(channel);

    ChannelFuture regFuture = config().group().register(channel);

    return regFuture;

}

public T newChannel() {

     return constructor.newInstance();

}

public NioServerSocketChannel(ServerSocketChannel channel) {

     super(null, channel, SelectionKey.OP_ACCEPT);

     config = new NioServerSocketChannelConfig(this, javaChannel().socket());

}

protected AbstractNioChannel(Channel parent, SelectableChannel ch, inteadInterestOp) {

     super(parent);

     this.ch = ch;

     this.readInterestOp = readInterestOp;

     ch.configureBlocking(false);

     }

 }

AbstractChannel(Channel parent) {

     this.parent = parent;

     id = newId();

     unsafe = newUnsafe();

     pipeline = newChannelPipeline();

 }

之后开始执行初始化,将bootstrapServer中的option和attr赋予serverSocketChannel,通过pipeline添加一个channelInitializer，进而通过channelInitializer将childOption、childAttr、workerGroup、childHandler保存在一个叫ServerBootstrapAcceptor的handler中。从名字可以猜测，此handler应该是在客户端连接时来处理相关事件的。而channelInitializer会在完成由子类实现的initChannel方法后将自己从pipeline中移除。

void init(Channel channel) {

    setChannelOptions(channel, options0().entrySet().toArray(newOptionArray(0)), logger);

    setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0)));

    ChannelPipeline p = channel.pipeline();

    final EventLoopGroup currentChildGroup = childGroup;

    final ChannelHandler currentChildHandler = childHandler;

    final Entry<ChannelOption<?>, Object>[] currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));

    final Entry<AttributeKey<?>, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));

    p.addLast(new ChannelInitializer<Channel>() {

    @Override

    public void initChannel(final Channel ch) {

        final ChannelPipeline pipeline = ch.pipeline();

        ChannelHandler handler = config.handler();

        if (handler != null) {

            pipeline.addLast(handler);

        }

        ch.eventLoop().execute(() -> pipeline.addLast(new ServerBootstrapAcceptor(

                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)));

    }

});

}

步骤3.2

初始化完毕后便是注册,首先判断是否注册以及传进来的eventLoop参数是否属于NioEventLoop类，然后判断当前线程是否是NioEventLoop线程，若是，则直接注册，否则添加到eventLoop的线程池中,通过创建nioEventLoop线程来执行注册。在register0——doRegister方法链中，我们看到netty最终调用了jdk底层的channel绑定了selector，由于此时还未绑定端口，所以ops即感兴趣事件是0。同时，把this，即NioServerSocketChannel作为attachment添加到selectionKey上，这是为了之后在select出事件时，可以获取channel进行操作。当注册完毕后，调用pipeline进行注册事件传播。如果设置了自动读取，还会立即开始一次读取。

// 通过unsafe对象进行注册

public ChannelFuture register(final ChannelPromise promise) {

    promise.channel().unsafe().register(this, promise);

    return promise;

}

// 如果当前线程是之前nioEventLoop绑定的线程则直接注册，否则添加到eventLoop的线程池中

public final void register(EventLoop eventLoop, final ChannelPromise promise) {

    AbstractChannel.this.eventLoop = eventLoop;

    if (eventLoop.inEventLoop()) {

        register0(promise);

    } else {

        eventLoop.execute(() -> register0(promise));

    }

}

// 注册后调用pipeline进行一些事件传播

private void register0(ChannelPromise promise) {

   boolean firstRegistration = neverRegistered;

   doRegister();

   neverRegistered = false;

   registered = true;

   pipeline.invokeHandlerAddedIfNeeded();

   pipeline.fireChannelRegistered();

   if (isActive()) {

       if (firstRegistration) {

           pipeline.fireChannelActive();

       } else if (config().isAutoRead()) {

           beginRead();

       }

   }

}

// 最终调用jdk底层channel进行注册

protected void doRegister() throws Exception {

    for (;;) {

        selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

        return;

    }

}

步骤3.3

回到AbstractBootStrap, 开始执行doBind0方法。这也是需要由nioEventLoop执行的，所以也丢到了线程池里。

private static void doBind0(final ChannelFuture regFuture, final Channel channel,

    final SocketAddress localAddress, final ChannelPromise promise) {

    channel.eventLoop().execute(new Runnable() {

        public void run() {

            if (regFuture.isSuccess()) {

                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

            } else {

                promise.setFailure(regFuture.cause());

            }

        }

    });

}

// 调用pipeline-tail进行绑定

public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {

    return pipeline.bind(localAddress, promise);

}

public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {

    return tail.bind(localAddress, promise);

}

@Override

public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {

    // 调用下一个OutBoundHandlerContext执行,默认传递到headContext

    final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);

    EventExecutor executor = next.executor();

    if (executor.inEventLoop()) {

        next.invokeBind(localAddress, promise);

    } else {

        safeExecute(executor, new Runnable() {

            @Override

            public void run() {

                next.invokeBind(localAddress, promise);

            }

        }, promise, null);

    }

    return promise;

}

// 传到headContext，调用unsafe来执行。

@Override

public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {

    unsafe.bind(localAddress, promise);

}

// AbstractUnsafe的bind

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {

    doBind(localAddress);

    if (!wasActive && isActive()) {

        invokeLater(new Runnable() {

        @Override

        public void run() {

            pipeline.fireChannelActive();

        }

    });

}

// unsafe最终调用jdk的channel完成绑定操作

protected void doBind(SocketAddress localAddress) throws Exception {

    javaChannel().bind(localAddress, config.getBacklog());

}

nioEventLoop通过pipeline的方式绑定,pipeline调用tailContext的bind方法，tailContext又会不断寻找下一个OutBoundHandler来执行bind方法，默认会传到headContext，headContext再调用底层的unsafe来执行bind。unsafe完成bind后，会通知pipeline调用fireChannelActive方法。这里绑定端口便完成了

步骤3.4

绑定端口后还需要注册感兴趣事件，这是通过fireChannelActive触发的。active事件首先会传递到headContext，

public void channelActive(ChannelHandlerContext ctx) {

    ctx.fireChannelActive();

    readIfIsAutoRead();

}

headContext继续将active事件传播，然后调用readIfIsAutoRead方法。此方法会调用channel的read方法，继而调用pipeline的read事件进行传播，又回到headContext，headContext又调用unsafe的beginRead如下:

protected void doBeginRead() throws Exception {

    final SelectionKey selectionKey = this.selectionKey;

    readPending = true;

    final int interestOps = selectionKey.interestOps();

    if ((interestOps & readInterestOp) == 0) {

        selectionKey.interestOps(interestOps | readInterestOp);

    }

}

在beginRead方法中,将channel注册感兴趣事件为创建NioServerSocketChannel时传入的OP_ACCEPT事件。

至此，初始化、注册selector、绑定、注册感兴趣事件4个小步骤都完成了

步骤4

主线程调用DefaultChannelPromise的sync方法，sync方法调用父类的await方法,阻塞在此处, 从此任由netty自由发挥，知道netty关闭了线程

步骤5

当netty报错或关闭后，主线程结束阻塞，开始执行netty优雅关闭机制。待续

巴特西