1. 首页
  2. Netty
  3. 正文

Netty源码解读（二）-服务端源码讲解

2022年6月16日 283点热度 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)

标签： 暂无
最后更新：2022年6月16日

王谷雨

一个苟且偷生的java程序员

文章评论

razz evil exclaim smile redface biggrin eek confused idea lol mad twisted rolleyes wink cool arrow neutral cry mrgreen drooling persevering

文章目录

COPYRIGHT © 2022 苟且偷生的java程序员. ALL RIGHTS RESERVED.

Theme Kratos Made By Seaton Jiang