Netty源码解读(二)-服务端源码讲解

2022年6月16日 92点热度 0人点赞 0条评论

简单Echo案例

注释版代码地址:netty

代码是netty的源码,我添加了自己理解的中文注释。

了解了Netty的线程模型和组件之后,我们先看看如何写一个简单的Echo案例,后续的源码讲解都基于此案例。以下是服务端的代码:

public final class MyEchoServer {

    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final MyEchoServerHandler serverHandler = new MyEchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    // 说明服务器端通道的实现类(便于 Netty 做反射处理)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    // 对服务端的 NioServerSocketChannel 添加 Handler
                    // LoggingHandler 是 netty 内置的一种 ChannelDuplexHandler,
                    // 既可以处理出站事件,又可以处理入站事件,即 LoggingHandler
                    // 既记录出站日志又记录入站日志。
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // 对服务端接收到的、与客户端之间建立的 SocketChannel 添加 Handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(serverHandler);
                        }
                    });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

EventLoopGroup的创建与初始化

对应代码

EventLoopGroup bossGroup = new NioEventLoopGroup();

默认线程数

跟踪NioEventLoopGroup的无参构造

NioEventLoopGroup()
-->
NioEventLoopGroup(int nThreads)
-->
NioEventLoopGroup(int nThreads, Executor executor)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider)
-->
NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory)
-->
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}

这里能看到,如果构造传入的线程数为0,则使用DEFAULT_EVENT_LOOP_THREADS

值为系统变量io.netty.eventLoopThreads,没有环境变量就取cpu逻辑线程数*2

例如我的电脑为8核16线程,nThreads = 16 * 2

EventLoop的创建

继续跟踪代码,以下代码有部分省略

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                        EventExecutorChooserFactory chooserFactory, Object... args) {
    // 检查线程数量不能小于1
    checkPositive(nThreads, "nThreads");

    // 这里的 ThreadPerTaskExecutor 实例是下文用于创建 EventExecutor 实例的参数
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }

    children = new EventExecutor[nThreads];

    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            // 创建EventLoop(重点)
            children[i] = newChild(executor, args);
            success = true;
        } catch (Exception e) {
            // TODO: Think about if this is a good exception type
            throw new IllegalStateException("failed to create a child event loop", e);
        } finally {
            。。。。。。
        }
    }

    // chooser 的作用是为了实现 next()方法,即从 EventLoopGroup 中挑选
    // 一个 NioEventLoop 来处理连接上 IO 事件的方法
    chooser = chooserFactory.newChooser(children);

    。。。。。。
}

ThreadPerTaskExecutor很简单,实现了Executor接口

public final class ThreadPerTaskExecutor implements Executor {
    。。。。。。
    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

这意味着每次执行executor.execute方法,都会开启一个线程。

EventLoop的创建是在newChild中

// 类NioEventLoopGroup
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    // selector工厂
    SelectorProvider selectorProvider = (SelectorProvider) args[0];
    // 选择策略工厂
    SelectStrategyFactory selectStrategyFactory = (SelectStrategyFactory) args[1];
    // 拒绝执行处理器(任务添加到队列中失败时调用)
    RejectedExecutionHandler rejectedExecutionHandler = (RejectedExecutionHandler) args[2];
    EventLoopTaskQueueFactory taskQueueFactory = null;
    EventLoopTaskQueueFactory tailTaskQueueFactory = null;

    int argsLength = args.length;
    if (argsLength > 3) {
        taskQueueFactory = (EventLoopTaskQueueFactory) args[3];
    }
    if (argsLength > 4) {
        tailTaskQueueFactory = (EventLoopTaskQueueFactory) args[4];
    }
    // 创建NioEventLoop并返回
    return new NioEventLoop(this, executor, selectorProvider,
            selectStrategyFactory.newSelectStrategy(),
            rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);
}

小结

NioEventLoopGroup的创建,初始化了selector工厂,选择策略,拒绝执行处理器等。

并创建了同样线程数的NioEventLoop

服务端引导类ServerBootstrap的创建与设置

对应代码

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

group的设置

public ServerBootstrap group(EventLoopGroup group) {
    return group(group, group);
}

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
    super.group(parentGroup);
    if (this.childGroup != null) {
        throw new IllegalStateException("childGroup set already");
    }
    this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
    return this;
}

提供了两个设置EventLoopGroup的方法,也就是parentGroup和childGroup可以是同一个group,

而parentGroup对应线程图中的bossGroup,childGroup对应线程图中的workerGroup

channel的设置

public B channel(Class<? extends C> channelClass) {
    return channelFactory(new <C>(
            ObjectUtil.checkNotNull(channelClass, "channelClass")
    ));
}

这里设置的是channel反射工厂,该工厂会使用反射生成NioServerSocketChannel对象。

创建并绑定channel

对应代码

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

channel的创建

准确点说,是负责创建连接(ACCEPT)的channel的创建

AbstractBootstrap#bind(int inetPort) -->
AbstractBootstrap#bind(SocketAddress localAddress) -->
private ChannelFuture doBind(final SocketAddress localAddress) {
    // 初始化 NioServerSocketChannel 的实例,并且将其注册到
    // bossGroup 中的 EvenLoop 中的 Selector 中,initAndRegister()

    // 实例的初始化和注册(此方法是异步的):
    // (1) 初始化:将handler注册进通道,并执行handler的handlerAdded、channelRegistered方法
    // (2) 将channel注册进selector
    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.
        // 若异步过程 initAndRegister()已经执行完毕,则进入该分支
        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.
        // 若异步过程 initAndRegister()还未执行完毕,则进入该分支
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            // 监听 regFuture 的完成事件,完成之后再调用
            // doBind0(regFuture, channel, localAddress, promise);
            @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;
    }
}

initAndRegister方法中主要做了3个操作,channel的创建、初始化以及将channel注册到EventLoop中

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 无参构造会创建pipelile
        // NioServerSocketChannel
        channel = channelFactory.newChannel();
        // 初始化相关属性
        // 如果是ServerBoottrap,还会设置bossGroup的handler,
        // 其中包括ServerBootstrap.handler设置的handler,以及最后添加ServerBootstrapAcceptor
        // ServerBootstrapAcceptor就是将channel注册到workerGroup的类
        init(channel);
    } catch (Throwable t) {
        。。。。。。
    }
    // 将channel注册进selector(监听ACCEPT事件)
    // 依然是通过开启eventLoop线程的方式进行注册
    // MultithreadEventLoopGroup
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}
  1. 创建

    请查看前面channel的设置一节,使用ReflectiveChannelFactory反射调用NioServerSocketChannel的无参构造器,创建channel

    NioServerSocketChannel() -->
    NioServerSocketChannel(SelectorProvider provider) -->
    public NioServerSocketChannel(SelectorProvider provider, InternetProtocolFamily family) {
        // newChannel(provider, family)生成Java NIO中的ServerSocketChannel
        this(newChannel(provider, family));
    }
    -->
    public NioServerSocketChannel(ServerSocketChannel channel) {
        // SelectionKey.OP_ACCEPT表示当前channel监听的是accept事件
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
    -->
    AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp)
    -->
    AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp)
    -->
    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        // 创建管道,同时创建头尾的handlerContext
        pipeline = newChannelPipeline();
    }

    我们看看管道的创建做了啥

    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);
    
        // 创建尾部HandlerContext
        tail = new TailContext(this);
        // 创建头部HandlerContext
        head = new HeadContext(this);
    
        // 初始化链条关系
        head.next = tail;
        tail.prev = head;
    }

    Pipeline的addLast实际上是插入,而不是在尾部添加。会将对应的handler封装成HandlerContext,插入到TailContext之前

    Untitled

    图片来源:netty源码分析之pipeline(一) - 简书 (jianshu.com)

    而在此案例中,pipeline长这样

    Untitled

  2. 初始化

    初始化调用的是init方法

    // ServerBootstrap类
    void init(Channel channel) {
        // 初始化相关属性
        setChannelOptions(channel, newOptionsArray(), logger);
        setAttributes(channel, newAttributesArray());
    
        ChannelPipeline p = channel.pipeline();
    
        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry, Object>[] currentChildOptions = newOptionsArray(childOptions);
        final Entry, Object>[] currentChildAttrs = newAttributesArray(childAttrs);
    
        p.addLast(new ChannelInitializer() {
            // 这里会在channel被注册进selector后执行
            @Override
            public void initChannel(final Channel ch) {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                // ServerBootstrap.handler(new LoggingHandler(LogLevel.INFO))
                // 就是在这里被设置进管道
                if (handler != null) {
                    pipeline.addLast(handler);
                }
    
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        // 负责监听READ的channel就是在这个handler中注册的
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

    初始化方法就是将ServerBootstrap一开始设置的相关属性初始化,以及往管道中添加handler,ServerBootstrapAcceptor这个handler是重点,我们在下面讲。

  3. 注册

    channel已经创建和初始化了,接下来就是将channel注册到EventLoop中

    // ChannelFuture regFuture = config().group().register(channel);
    // 类MultithreadEventLoopGroup
    public ChannelFuture register(Channel channel) {
        // next():从EventLoopGroup中选择一个EventLoop
        return next().register(channel);
    }
    -->
    SingleThreadEventLoop#register(Channel channel)
    -->
    SingleThreadEventLoop#register(final ChannelPromise promise)
    -->
    // 类AbstractChannel
    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        。。。。。。
    
        AbstractChannel.this.eventLoop = eventLoop;
    
        // 判断如果当前线程是EventLoop的线程,则直接执行
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            // 否则添加进EventLoop的任务队列,由EventLoop的线程去执行
            try {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                。。。。。。
            }
        }
    }

    register0方法就是注册方法了,至于为什么会有eventLoop.execute,这个方法很有意思,等会讲。

    先看看register0

    private void register0(ChannelPromise promise) {
        try {
            。。。。。。
            boolean firstRegistration = neverRegistered;
            // 将当前channel注册到selector中
            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.
            // 触发channel的handlerAdded方法
            // 如果是ChannelInitializer的话,会在handlerAdded中触发initChannel方法
            pipeline.invokeHandlerAddedIfNeeded();
    
            safeSetSuccess(promise);
            // 触发channel的channelRegistered方法
            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) {
            。。。。。。
        }
    }

    doRegister():将当前channel注册到selector中,里面会调用到Java NIO的一些API
    pipeline.invokeHandlerAddedIfNeeded():触发channel的handlerAdded方法
    pipeline.fireChannelRegistered():触发channel的channelRegistered方法

    从这里可以看到,全部handler中的handlerAdded执行完,才会执行channelRegistered方法

绑定端口

重新回到AbstractBootstrap#doBind方法中

// 如果上面的initAndRegister方法执行完毕(异步执行的),则执行doBind0
if (regFuture.isDone()) {
    // At this point we know that the registration was complete and successful.
    // 若异步过程 initAndRegister()已经执行完毕,则进入该分支
    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.
    // 若异步过程 initAndRegister()还未执行完毕,则进入该分支
    final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
    regFuture.addListener(new ChannelFutureListener() {
        // 监听 regFuture 的完成事件,完成之后再调用
        // doBind0(regFuture, channel, localAddress, promise);
        @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;
}

上面这段if/esle做了同一件是,就是自行doBind0方法,区别在于如果initAndRegister执行完毕,则执行调用doBind0,否则添加监听器,等执行完成触发调用doBind0

继续看doBind0

// 类AbstractBootstrap
private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    // execute方法会将这个Runnable加入到taskQueue中,并开线程执行EventLoop的run方法(死循环)
    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());
            }
        }
    });
}

channel.eventLoop().execute这个后面再说,可以看到,里面的逻辑是调用channel.bind在实现绑定的,继续跟踪

AbstractChannel#bind(SocketAddress localAddress, ChannelPromise promise)
-->
// 类AbstractChannel
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
    // tail就是TailContext
    return tail.bind(localAddress, promise);
}
-->
// 类AbstractChannel
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
    ObjectUtil.checkNotNull(localAddress, "localAddress");
    if (isNotValidPromise(promise, false)) {
        // cancelled
        return promise;
    }
    // 在管道中从当前handlerContext往前查找实现了bind方法的handlerContext
    final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        // 执行handlerContext的bind方法
        next.invokeBind(localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeBind(localAddress, promise);
            }
        }, promise, null, false);
    }
    return promise;
}
-->
// 类AbstractChannel
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);
    }
}

从上面可以看到,最终会执行handler的bind方法,拿LoggingHandler的bind方法举例

// 类LoggingHandler
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
    if (logger.isEnabled(internalLevel)) {
        logger.log(internalLevel, format(ctx, "BIND", localAddress));
    }
    ctx.bind(localAddress, promise);
}

ctx.bind(localAddress, promise)是不是很眼熟,没错,就是AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)

就像一个循环,每一次都在当前handlerContext往前找有实现了bind方法的handlerContext,执行bind,然后继续往前找。

最终找到管道中的第一个handler,也就是HeadContext,看看它实现的bind方法

// 类HeadContext
public void bind(
        ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
    unsafe.bind(localAddress, promise);
}
-->
AbstractChannel#bind(final SocketAddress localAddress, final ChannelPromise promise)
-->
NioServerSocketChannel#doBind(SocketAddress localAddress)

最后,还是Java NIO的API来绑定


参考资料:

《Netty in Action》,Norman Maurer

《Scalable IO in Java》,Doug Lea

45 张图深度解析 Netty 架构与原理 (qq.com)

王谷雨

一个苟且偷生的java程序员

文章评论