示例代码

我们就以下面这段代码为例，来分析一个Netty是如何启动的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class HttpServer {

    public static void main(String[] args) throws InterruptedException {


        //负责接收客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        //负责网络读写
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    //http编解码器
                    pipeline.addLast(new HttpServerCodec());
                    //自定义的handler
                    pipeline.addLast(new HttpServerHandler());
                }
            });

            final ChannelFuture future = bootstrap.bind(3333).sync();
            future.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }


    }
}

这段代码我们首先创建了两个EventLoopGroup。然后创建了ServerBootstrap并设置设置了一系列的属性。然后调用了bind方法完成了服务器的启动。所以要分析Netty服务端的启动过程就需要从bind方法入手。

启动流程分析

从bind方法来查看整个启动的流程：

1
2
3
4
5
6
7
8
9
10
public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
}

public ChannelFuture bind(SocketAddress localAddress) {
/*启动前的检测，检测EventLoopGroup是否已经设置，
ChannelFactory是否已经设置。否则抛出异常*/
    validate();
    return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
}

bind方法，首先进行了启动前的检查，然后调用doBind方法开始真正的服务端启动。
那么，下面我们来看doBind方法的具体实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

private ChannelFuture doBind(final SocketAddress localAddress) {
    //初始化并注册一个Channel，
    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进行绑定
        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;
    }
}

这个方法主要做了这些事情：

  1. 调用initAndRegister()拿到一个ChannelFuture对象regFuture
  2. 根据regFuture判断该对象是否抛出异常，如果是，直接返回
  3. 根据regFuture判断initAndRegister是否执行完毕，如果执行完毕，则调用doBind0
  4. 如果initAndRegister没有执行完毕，就会对regFuture对象添加一个ChannelFutureListener来监听initAndRegister执行完毕的事件，一旦执行完毕，就会和2，3步骤一样进行异常判断和调用doBind0。

看完这个方法的实现，我们有两个大大的疑问，initAndRegister()到底干了什么事情，doBind0又干了什么事情。
下面我们就来探究这两个问题：

我们首先来看一看initAndRegister方法的底层实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
    //新建一个Channel
        channel = channelFactory.newChannel();
        //初始化新建的Channel
        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);
    }

    //向EventLoopGroup中注册一个channel
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

整个方法主要完成两件事情：

  1. 新建一个channel
  2. 向EventLoopGroup中注册channel。

我们在初始化ServerBootstrap的时候有这么一行代码 .channel(NioServerSocketChannel.class)
新建channe就是利用反射实例化了我们设置的channel类。

注册channel的过程也值得我们注意：
ChannelFuture regFuture = config().group().register(channel);
group()方法返回的是前面的boss NioEvenLoopGroup,它继承自MultithreadEventLoopGroup,这里所调用的register也是MultithreadEventLoopGroup 中的方法。
具体实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public ChannelFuture register(Channel channel) {
    /*next方法从EventLoopGroup中获取下一个EventLoop*/
    return next().register(channel);
}

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

public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    //拿到channel的unsafe对象，进行注册
    promise.channel().unsafe().register(this, promise);
    return promise;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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;

    //确保注册是由当前的EventLoop发起的
    if (eventLoop.inEventLoop()) {
    //注册
        register0(promise);
    } else {
    //如果注册不是当前EventLoop发起的，
    //就封装成一个task异步执行
        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);
        }
    }
}

这个方法的过程如下：

  1. 首先判断该channel是否已经注册到EventLoop中
  2. 判断当前当前线程是否未该EventLoop发起的，如果是，则调用register0直接进行注册。
  3. 如果不是，则说明该EventLoop中的线程此时没有执行全，则需要新建一个线程，单独封装一个task，而这个task的主要任务也是调用register0

下面我们就来分析一下register0的具体实现；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
private void register0(ChannelPromise promise) {
        try {
           //确保channel处于open状态
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            boolean firstRegistration = neverRegistered;
            //真正的注册
            doRegister();
            neverRegistered = false;
            registered = true;

            pipeline.invokeHandlerAddedIfNeeded();
            //设置注册的结果为成功
            safeSetSuccess(promise);
            //如果是首次注册，发起pipeline的fireChannelActive
            pipeline.fireChannelRegistered();
            
            if (isActive()) {
                if (firstRegistration) {
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    beginRead();
                }
            }
        } catch (Throwable t) {
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

这个方法的工作流程如下：

  1. 如果Channel处于open状态，则调用doRegister完成注册，然后将注册结果设置为成功
  2. 如果是首次注册且处于激活状态，则发起pipeline的fireChannelActive()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
protected void doRegister() throws Exception {
     boolean selected = false;
     for (;;) {
         try {
         
         //注册到NIOEventloop的Selector上
             selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
             return;
         } catch (CancelledKeyException e) {
             if (!selected) {
                 eventLoop().selectNow();
                 selected = true;
             } else {
                 throw e;
             }
         }
     }
 }

到此，我们基本清楚了initAndRegister()方法到底做了什么事情。
接下来，我们继续分析doBind0方法的具体实现；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {
        //新建一个任务，然后提交给EventLoop
    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());
            }
        }
    });
}

这个方法的主要逻辑还是非常清晰的。我们首先来分析一下execute的实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }

    boolean inEventLoop = inEventLoop();
    //添加任务到队列中，如果入队失败，则执行拒绝策略
    addTask(task);
    if (!inEventLoop) {
        startThread();
        if (isShutdown()) {
            boolean reject = false;
            try {
                if (removeTask(task)) {
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
            }
            if (reject) {
                reject();
            }
        }
    }

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

task添加到任务队列成功后，执行任务会调用如下方法:
channel.bind(localAddress,promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

channel首先调用bind完成channel完成与端口的绑定：

1
2
3
4
5
6
7
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);
}

bind方法最终会调用DefaultChannelPipeline 的bind方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
    //进行参数校验
    if (localAddress == null) {
        throw new NullPointerException("localAddress");
    }
    if (!validatePromise(promise, false)) {
        return promise;
    }

//从 AbstractChannelHandlerContext 
//双向链表尾部开始遍历找到第一个节点属性为outbound为true的节点
    final AbstractChannelHandlerContext next = findContextOutbound();
    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;
}

invokeBind的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
    if (invokeHandler()) {
        try {
            ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    } else {
        bind(localAddress, promise);
    }
}

headler()返回的是HeadContext对象，然后调用bind()方法：bind方法的具体实现如下：

1
2
3
4
5
6
public void bind(
        ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
        throws Exception {
    unsafe.bind(localAddress, promise);
}

最终是调用了unsafe的bind方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    assertEventLoop();

    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        return;
    }

    if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
        localAddress instanceof InetSocketAddress &&
        !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
        !PlatformDependent.isWindows() && !PlatformDependent.isRoot()) {

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

内部又调用了doBind方法，它绑定的核心方法。

1
2
3
4
5
6
7
8
protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}

javaChannel方法返回的是NioServerSocketChannel 实例初始化时所产生的 Java NIO ServerSocketChannel实例（ServerSocketChannelImple实例），然后调用其 bind()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public ServerSocketChannel bind(SocketAddress var1, int var2) throws IOException {
    Object var3 = this.lock;
    synchronized(this.lock) {
        if(!this.isOpen()) {
            throw new ClosedChannelException();
        } else if(this.isBound()) {
            throw new AlreadyBoundException();
        } else {
            InetSocketAddress var4 = var1 == null?new InetSocketAddress(0):Net.checkAddress(var1);
            SecurityManager var5 = System.getSecurityManager();
            if(var5 != null) {
                var5.checkListen(var4.getPort());
            }

            NetHooks.beforeTcpBind(this.fd, var4.getAddress(), var4.getPort());
            Net.bind(this.fd, var4.getAddress(), var4.getPort());
            Net.listen(this.fd, var2 < 1?50:var2);
            Object var6 = this.stateLock;
            synchronized(this.stateLock) {
                this.localAddress = Net.localAddress(this.fd);
            }

            return this;
        }
    }
}

这个方法就属于NIO层次的了。捯饬就完成了服务端端口的绑定工作。

技术Netty
许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

快来参与讨论吧~