剖析Netty內部網絡實現原理

2022.07.14
剖析Netty內部網絡實現原理

Netty 網絡模塊的工作核心原理咱們就介紹完了。飛哥一直“鼓吹”內功的好處。只要你具備了堅實的內功，各種語言裡看似風牛馬不相及的東西，在底層實際上原理是想通的。我本人從來沒用Java 開發過服務器程序，更沒碰過Netty。

Netty 是一個在Java 生態裡應用非常廣泛的的網絡編程工具包，它在2004 年誕生到現在依然是火的一塌糊塗，光在github 上就有30000 多個項目在用它。所以要想更好地掌握網絡編程，我想就繞不開Netty。所以今天我們就來分析分析Netty 內部網絡模塊的工作原理。

友情提示，本文算上代碼將近有兩三萬字，比較長，如果時間緊迫中間部分可以跳著看。第一節和最後的第六節建議必讀。當然直接拖到尾部收藏點贊點轉發，也是ok 的，哈哈！

另外，今天又給大家申請到了贊助，在文末給大家申請了5本冰河的新書《深入理解高並發編程》，抽獎送給大家。

一、Netty 用法

我們首先找一個Netty 的例子，本篇文章整體都是圍繞這個例子來展開敘述的。我們下載Netty 的源碼，並在examples 中找到echo 這個demo。同時，為了防止代碼更新導致對本文敘述的影響，我們切到4.1 分支上來。

# git checkout https://github.com/netty/netty.git
# git checkout -b 4.1
# cd example/src/main/java/io/netty/example/echo
  • 1.
  • 2.
  • 3.

在這個demo 的EchoServer 中，展示了使用Netty 寫Server 的經典用法。（飛哥在文章中會在不影響核心邏輯的表達上，對原始代碼盡心適當的精簡，比如下面代碼中的try 就被我丟了）

public final class EchoServer {
 public static void main(String[] args) throws Exception {
  EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();

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

        // Start the server.
        ChannelFuture f = b.bind(PORT).sync();
        ......
 }
}
  • 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.

如果你是一個Java 新手，或者乾脆像飛哥一樣沒用Netty 寫過服務，相信上述代碼基本是看不懂的。究其根本原因是相比C/C++ ，Java 的封裝程度比較高。Java 語言本身的JVM 中NIO 對網絡的封裝就已經屏蔽了很多底層的概念了，再加上Netty 又封裝了一層，所以Java 開發者常用的一些術語和概念和其它語言出入很大。

比如上面代碼中的Channel、NioEventLoopGroup 等都是其它語言中所沒見過的。不過你也不用感到害怕，因為這其中的每一個概念都是socket、進程等底層概念穿了一身不同的衣服而已。接下來我們分別細了解一下這些概念。

1.1 NioEventLoopGroup

如果你沒接觸過Netty，可以簡單把NioEventLoopGroup 理解為一個線程池就可以。每一個NioEventLoopGroup 內部包含一個或者多個NioEventLoop。

圖片

其中 NioEventLoop 是对线程、epoll 等概念进行了一个集中的封装。

首先，EventLoop 本身就是一个线程。为什么这么说，我们通过看 NioEventLoop 的继承关系就能看出来。NioEventLoop 继承于 SingleThreadEventLoop，而 SingleThreadEventLoop 又继承于 SingleThreadEventExecutor。SingleThreadEventExecutor 实现了在 Netty 中对本地线程的抽象。

public abstract class SingleThreadEventExecutor extends ... {
 private volatile Thread thread;
 private final Queue<Runnable> taskQueue;
}
  • 1.
  • 2.
  • 3.
  • 4.

在 SingleThreadEventExecutor 中不但封装了线程对象 Thread，而且还配置了一个任务队列 taskQueue，用于其它线程向它来放置待处理的任务。

1.2 selector

另外 NioEventLoopEventLoop 以 selector 的名义封装了 epoll（在 Linux 操作系统下）。

圖片

在 NioEventLoop 对象内部，会有 selector 成员定义。这其实就是封装的 epoll 而来的。我们来看具体的封装过程。以及 selectedKeys，这是从 selector 上发现的待处理的事件列表。

public final class NioEventLoop extends SingleThreadEventLoop{
 // selector
 private Selector selector;
    private Selector unwrappedSelector;

    // selector 上发现的各种待处理事件
    private SelectedSelectionKeySet selectedKeys;
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

NioEventLoopGroup 在构造的时候，会调用 SelectorProvider#provider 来生成 provider，在默认情况下会调用 sun.nio.ch.DefaultSelectorProvider.create 来创建。

//file:java/nio/channels/spi/SelectorProvider.java
public abstract class SelectorProvider {

    public static SelectorProvider provider() {
     // 1. java.nio.channels.spi.SelectorProvider 属性指定实现类
        // 2. SPI 指定实现类
        ......

        // 3. 默认实现，Windows 和 Linux 下不同
        provider = sun.nio.ch.DefaultSelectorProvider.create();
        return provider;
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.

在 Linux 下，默认创建的 provider 使用的就是 epoll。

//file:sun/nio/ch/DefaultSelectorProvider.java
public class DefaultSelectorProvider {
 public static SelectorProvider create() {
        String osname = AccessController
            .doPrivileged(new GetPropertyAction("os.name"));
        if (osname.equals("Linux"))
            return createProvider("sun.nio.ch.EPollSelectorProvider");
    }

}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.

1.3 Channel

Channel 是 JavaNIO 里的一个概念。大家把它理解成 socket，以及在 socket 之上的一系列操作方法的封装就可以了。

圖片

Java 在 Channel 中把 connect、bind、read、write 等方法都以成员方法的形式给封装起来了。

public interface Channel extends ... {
 Channel read();
 Channel flush();
 ......

 interface Unsafe {
  void bind(SocketAddress localAddress, ...);
  void connect(SocketAddress remoteAddress, ...);
  void write(Object msg, ...);
  ......
 }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.

另外在 Java 中，习惯把 listen socket 叫做父 channel，客户端握手请求到达以后创建出来的新连接叫做子 channel，方便区分。

1.4 Pipeline

在每个 Channel 对象的内部，除了封装了 socket 以外，还都一个特殊的数据结构 DefaultChannelPipeline pipeline。在这个 pipeline 里是各种时机里注册的 handler。

Channel 上的读写操作都会走到这个 DefaultChannelPipeline 中，当 channel 上完成 register、active、read、readComplete 等操作时，会触发 pipeline 中的相应方法。

圖片

这个 ChannelPipeline 其实就是一个双向链表，以及链表上的各式各样的操作方法。

public interface ChannelPipeline {
 ChannelPipeline addFirst(String name, ChannelHandler handler);
 ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler);
 ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler);
 ChannelPipeline addLast(String name, ChannelHandler handler);

 ChannelPipeline fireChannelRead(Object msg);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

1.5 EchoServer 解读

现在我们具备了对 Java、对 Netty 的初步理解以后，我们再会后来看一下开篇提到的 EchoServer 源码。

public final class EchoServer {
 public static void main(String[] args) throws Exception {
  EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();

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

        // Start the server.
        ChannelFuture f = b.bind(PORT).sync();
        ......
 }
}
  • 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.

在该代码一开头，bossGroup = new NioEventLoopGroup(1) 这一行是创建了一个只有一个线程的线程池。workerGroup = new NioEventLoopGroup 又创建了 worker 线程池，没有指定数量，Netty 内部会根据当前机器的 CPU 核数来灵活决定。

ServerBootstrap 这是一个脚手架类，是为了让我们写起服务器程序来更方便一些。

b.group(bossGroup, workerGroup) 这一行是将两个线程池传入，第一个作为 boss 只处理 accept 接收新的客户端连接请求。第二个参数作为 worker 线程池，来处理连接上的请求接收、处理以及结果发送发送。

我们注意下 childHandler是传入了一个 ChannelInitializer，这是当有新的客户端连接到达时会回调的一个方法。在这个方法内部，我们给这个新的 chaneel 的 pipeline 上添加了一个处理器 serverHandler，以便收到数据的时候执行该处理器进行请求处理。

上面的几个方法都是定义，在 b.bind 方法中真正开始启动服务，创建父 channel（listen socket），创建 boss 线程。当有新连接到达的时候 boss 线程再创建子 channel，为其 pipeline 添加处理器，并启动 worker 线程来进行处理。

二、Netty bootstrap 参数构建

简言之 bootstrap.group() .channel() .childHandler() .childOption() 就是在构建 Netty Server 的各种参数。

2.1 group 设置

ServerBootstrap 和其父类 AbstractBootstrap 内部分别定义了两个 EventLoopGroup group 成员。父类 AbstractBootstrap 的 group 是用来处理 accpet 事件的，ServerBootstrap 下的 childGroup 用来处理其它所有的读写等事件。

group() 方法就是把 EventLoopGroup 参数设置到自己的成员上完事。其中如果调用 group() 只传入了一个线程池，那么将来本服务下的所有事件都由这个线程池来处理。详情查看飞哥精简后的源码。

//file:io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
 //用来处理非 accept 以外的线程池
 private volatile EventLoopGroup childGroup;

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

 public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        super.group(parentGroup);
        this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
        return this;
    }
}

public abstract class AbstractBootstrap ... {
 //用来处理 accept 的线程
 volatile EventLoopGroup group;

 public B group(EventLoopGroup group) {
        this.group = group;
        ......
    }
}
  • 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.

2.2 channel 设置

再看 ServerBootstrap#channel 方法 是用来定义一个工厂方法，将来需要创建 channel 的时候都调用该工厂进行创建。

//file:io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

 public B channel(Class<? extends C> channelClass) {
        return channelFactory(new ReflectiveChannelFactory<C>(
                ObjectUtil.checkNotNull(channelClass, "channelClass")
        ));
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

回头看本文开头 demo，.channel(NioServerSocketChannel.class) 指的是将来需要创建 channel 的时候，创建 NioServerSocketChannel 这个类型的。

2.3 option 设置

再看 option 方法，只是设置到了 options 成员中而已

//file:io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

    public <T> B option(ChannelOption<T> option, T value) {
        ObjectUtil.checkNotNull(option, "option");
        synchronized (options) {
            if (value == null) {
                options.remove(option);
            } else {
                options.put(option, value);
            }
        }
        return self();
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.

2.4 handler 方法

本文 demo 设置了两处 handler，一处是 handler，另一处是 childHandler。他们都是分别设置到自己的成员上就完事，看源码。

//file:io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends ...... {

    public B handler(ChannelHandler handler) {
        this.handler = ObjectUtil.checkNotNull(handler, "handler");
        return self();
    }
    public ServerBootstrap childHandler(ChannelHandler childHandler) {
        this.childHandler = ObjectUtil.checkNotNull(childHandler, "childHandler");
        return this;
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.

三、Netty bootstrap 启动服务

ServerBootstrap 下的 bind 方法是服务启动过程中非常重要的一个方法。创建父 channel（listen socket），创建 boss 线程，为 boss 线程绑定 Acceptor 处理器，调用系统调用 bind 进行绑定和监听都是在这里完成的。

先来直接看一下 bind 相关的入口源码。

//file:io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap ... {
 ......
}

//file:io/netty/bootstrap/AbstractBootstrap.java
public abstract class AbstractBootstrap ... {

    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        return doBind(...);
    }

    private ChannelFuture doBind(final SocketAddress localAddress) {

     //创建父 channel、初始化并且注册
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        ......

        //如果 Register 已经完成，则直接 doBind0
        if (regFuture.isDone()) {
         ChannelPromise promise = channel.newPromise();
         doBind0(regFuture, channel, localAddress, promise);
         return promise;
        //否则就注册一个 listener（回调），等 register 完成的时候调用 
        } else {
         final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
             regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                 promise.registered();
                    doBind0(regFuture, channel, localAddress, promise);
                }
            return promise;
        }
        
    }

    //创建 channel，对其初始化，并且 register（会创建 parent 线程）
    final ChannelFuture initAndRegister() {
     //3.1 创建父 channel（listen socket）
        channel = channelFactory.newChannel();

        //3.2 对父 channel（listen socket）进行初始化
        init(channel);

        //3.3 注册并启动 boss 线程
        ChannelFuture regFuture = config().group().register(channel);
        ......
    }

    //3.4 真正的bind
    private static void doBind0(...) {
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                ......
            }
        });
    }
}
  • 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.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.

在这个过程中，做了如下几件重要的事情

  • 创建父 channel（listen socket）
  • 对父 channel（listen socket）进行初始化
  • register父 channel（listen socket）到主 group，并启动主进程
  • 真正的 bind

接下来我们分开来看。

3.1 创建父 channel（listen socket）

在 initAndRegister() 方法中创建 channel（socket），它调用了 channelFactory.newChannel()。

public abstract class AbstractBootstrap
    //创建 channel，对其初始化，并且 register（会创建 parent 线程）
    final ChannelFuture initAndRegister() {
     //3.1 创建 listen socket
        channel = channelFactory.newChannel();
        ......
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

回想下 2.2 节的channel 方法，返回的是一个反射 ReflectiveChannelFactory。没错这里的 newChannel 就是调用这个工厂方法来创建出来一个 NioServerSocketChannel 对象。

3.2 对父 channel（listen socket）进行初始化

在 initAndRegister 创建除了 channel 之后，需要调用 init 对其进行初始化。

public abstract class AbstractBootstrap
    final ChannelFuture initAndRegister() {
     //3.1 创建父 channel（listen socket）
        //3.2 对父 channel（listen socket）进行初始化
        init(channel);
        ......
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

在 init() 中对 channel 进行初始化，一是给 options 和 attrs 赋值，二是构建了父 channel 的 pipeline。

//file:src/main/java/io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

 void init(Channel channel) {
  //设置 option 和 attr
        setChannelOptions(channel, newOptionsArray(), logger);
        setAttributes(channel, newAttributesArray());

        //设置 pipeline
        ChannelPipeline p = channel.pipeline();
        p.addLast(new ChannelInitializer<Channel>() {
           ......
        });
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.

在 setChannelOptions 中对 channel 的各种 option 进行设置。回忆我们在使用 ServerBootstrap 时可以传入 SO_BACKLOG，这就是其中的一个 option。在这里会真正设置到 channel（socket）上。

ServerBootstrap b = new ServerBootstrap();
b.option(ChannelOption.SO_BACKLOG, 100)
  • 1.
  • 2.

在 init 中，稍微难理解一点是 p.addLast(new ChannelInitializer...)。这一段代码只是给父 channel 添加一个 handler 而已。其真正的执行要等到 register 后，我们待会再看。

3.3 register 父 channel

父 channel 在创建完，并且初始化之后，需要注册到 boss 线程上才可用。

public abstract class AbstractBootstrap
    final ChannelFuture initAndRegister() {
     //3.1 创建父 channel（listen socket）
        //3.2 对父 channel（listen socket）进行初始化
        //3.3 注册并启动 boss 线程
        ChannelFuture regFuture = config().group().register(channel);
        ......
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

其中 config().group() 最终会调用到 AbstractBootstrap#group，在这个方法里获取的是我们传入进来的 bossGroup。

public abstract class AbstractBootstrap
    volatile EventLoopGroup group;
    public final EventLoopGroup group() {
        return group;
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.

其中 bossGroup 是一个 NioEventLoopGroup 实例，所以代码会进入到 NioEventLoopGroup#register 方法。

public class NioEventLoopGroup extends MultithreadEventLoopGroup {}

public abstract class MultithreadEventLoopGroup extends ... {

 @Override
    public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }

    @Override
    public EventLoop next() {
        return (EventLoop) super.next();
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.

在 NioEventLoopGroup 里包含一个或多个 EventLoop。上面的 next 方法就是从中选择一个出来，然后将 channel 注册到其上。

对于本文来讲，我们使用的是 NioEventLoopGroup，其内部包含的自然也就是 NioEventLoop，我们继续查找其 register 方法。

public final class NioEventLoop extends SingleThreadEventLoop
 //在 eventloop 里注册一个 channle（socket）
 public void register(final SelectableChannel ch, ...) {
  ......
        register0(ch, interestOps, task);
 }

 //最终调用 channel 的 register
 private void register0(SelectableChannel ch, int interestOps, NioTask<?> task) {
        ch.register(unwrappedSelector, interestOps, task);
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.

可见，NioEventLoop 的 register 最后又调用到 channel 的 register 上了。在我们本文中，我们创建的 channel 是 NioServerSocketChannel，我们就依照这条线索来查。

//file:src/main/java/io/netty/channel/AbstractChannel.java
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {

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

        //关联自己到 eventLoop
        AbstractChannel.this.eventLoop = eventLoop;

        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
         try {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(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.

在 channel 的父类 AbstractChannel 中的 register 中，先是把自己关联到传入的 eventLoop 上。接着调用 inEventLoop 来判断线程当前运行的线程是否是 EventExecutor的支撑线程，是则返回直接 register0。

一般来说，服务在启动的时候都是主线程在运行。这个时候很可能 boss 线程还没有启动。所以如果发现当前不是 boss 线程的话，就调用 eventLoop.execute 来启动 boss 线程。

NioEventLoop 的父类是 SingleThreadEventExecutor, 找到 execute 方法。

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {

    public void execute(Runnable task) {
        execute0(task);
    }

    private void execute0(@Schedule Runnable task) {
        execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));
    }

    private void execute(Runnable task, boolean immediate) {
        boolean inEventLoop = inEventLoop();
        addTask(task);
        if (!inEventLoop) {
            startThread();
        }

        if (!addTaskWakesUp && immediate) {
            wakeup(inEventLoop);
        }
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.

我们先来看 addTask(task)，它是将 task 添加到任务队列中。等待线程起来以后再运行。

public abstract class SingleThreadEventExecutor extends ... {

 private final Queue<Runnable> taskQueue;
 protected void addTask(Runnable task) {
        (task);
    }
 final boolean offerTask(Runnable task) {
        return taskQueue.offer(task);
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.

inEventLoop() 是判断当前线程是不是自己绑定的线程，这时还在主线程中运行，所以 inEventLoop 为假，会进入 startThread 开始为 EventLoop 创建线程。

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {

    private void startThread() {
        doStartThread();
        ......
    }

    private void doStartThread() {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                SingleThreadEventExecutor.this.run();
                ......
            }
        }
    }  
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.

在 doStartThread 中调用 Java 线程管理工具 Executor 来启动 boss 线程。

3.4 boss 线程启动

当线程起来以后就进入了自己的线程循环中了，会遍历自己的任务队列，然后开始处理自己的任务。

public final class NioEventLoop extends SingleThreadEventLoop {

    protected void run() {
        for (;;) {
            if (!hasTasks()) {
                strategy = select(curDeadlineNanos);
            }

            //如果有任务的话就开始处理
            runAllTasks(0);

            //任务处理完毕就调用 epoll_wait 等待事件发生
            processSelectedKeys();
        }
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.

前面我们在 3.3 节看到 eventLoop.execute 把一个 Runnable 任务添加到了任务队列里。当 EventLoop 线程启动后，它会遍历自己的任务队列并开始处理。这时会进入到 AbstractChannel#register0 方法开始运行。

//file:src/main/java/io/netty/channel/AbstractChannel.java
public abstract class AbstractChannel extends ... {

    public final void register(...) {
        eventLoop.execute(new Runnable() {
            @Override
            public void run() {
                register0(promise);
            }
        });
        ......
    }
    private void register0(ChannelPromise promise) {
        doRegister();
        ......
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.

函数 doRegister 是在 AbstractNioChannel 类下。

//file:io/netty/channel/nio/AbstractNioChannel.java
public abstract class AbstractNioChannel extends AbstractChannel {

    private final SelectableChannel ch;

    protected SelectableChannel javaChannel() {
        return ch;
    }
    public NioEventLoop eventLoop() {
        return (NioEventLoop) super.eventLoop();
    }
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);            
        }
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.

上面最关键的一句是 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);。这一句就相当于在 C 语言下调用 epoll_ctl 把 listen socket 添加到了 epoll 对象下。

其中 javaChannel 获取父 channel，相当于 listen socket。unwrappedSelector 获取 selector，相当于 epoll 对象。register 相当于使用 epoll_ctl 执行 add 操作。

当 channel 注册完后，前面 init 时注册的 ChannelInitializer 回调就会被执行。再回头看它的 回调定义。

//file:src/main/java/io/netty/bootstrap/ServerBootstrap.java
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

 void init(Channel channel) 
  ......

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) {
             ......
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.

在 ChannelInitializer#initChannel 里，又给 boss 线程的 pipeline 里添加了一个任务。该任务是让其在自己的 pipeline 上注册一个 ServerBootstrapAcceptor handler。将来有新连接到达的时候，ServerBootstrapAcceptor 将会被执行。

3.5 真正的 bind

再看 doBind0 方法，调用 channel.bind 完成绑定。

private static void doBind0(...) {
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                ......
            }
        });
    }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

四、新连接到达

我们再回到 boss 线程的主循环中。

public final class NioEventLoop extends SingleThreadEventLoop {
    protected void run() {
        for (;;) {
            strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());

            //任务队列都处理完就开始 select
            if (!hasTasks()) {
                strategy = select(curDeadlineNanos);
            }
            //处理各种事件
            if (strategy > 0) {
                processSelectedKeys();
            }
        }  
    }
    
    private int select(long deadlineNanos) throws IOException {
        if (deadlineNanos == NONE) {
            return selector.select();
        }
        // Timeout will only be 0 if deadline is within 5 microsecs
        long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
        return timeoutMillis <= 0 ? selector.selectNow() : selector.select(timeoutMillis);
    }
}
  • 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.

假如线程任务队列中的任务都处理干净了的情况下，boss 线程会调用 select 来发现其 selector 上的各种事件。相当于 C 语言中的 epoll_wait。

当发现有事件发生的时候，例如 OP_WRITE、OP_ACCEPT、OP_READ 等的时候，会进入相应的处理

public final class NioEventLoop extends SingleThreadEventLoop {
    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
     ......
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
         ch.unsafe().forceFlush();
     }
     if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
         unsafe.read();
     }
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.

对于服务端的 Unsafe.read()  这里会执行 io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe.read() 方法，它会调用 JDK 底层的 ServerSocketChannel.accept() 接收到客户端的连接后，将其封装成 Netty 的 NioSocketChannel，再通过 Pipeline 将 ChannelRead 事件传播出去，这样 ServerBootstrapAcceptor 就可以在 ChannelRead 回调里处理新的客户端连接了。

我们直接看 ServerBootstrapAcceptor#ChannelRead。

//file:
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
 
 ......

    private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {

        public void channelRead(ChannelHandlerContext ctx, Object msg) {
         // 获取child channel
            final Channel child = (Channel) msg;

            // 设置 childHandler 到 child channel
            child.pipeline().addLast(childHandler);

            // 设置 childOptions、 childAttrs
            setChannelOptions(child, childOptions, logger);
            setAttributes(child, childAttrs);

            // 将 child channel 注册到 childGroup
            childGroup.register(child).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        forceClose(child, future.cause());
                    }
                }
            });
        }
    }
}
  • 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.

在 channelRead 先是获取到了新创建出来的子 channel，并为其 pipeline 添加 childHandler。回头看 1.5 节，childHandler 是我们自定义的。

紧接着调用 childGroup.register(child) 将子 channel 注册到 workerGroup 上。这个 register 过程和 3.3节、3.5节过程一样。区别就是前面是父 channel 注册到 bossGroup 上，这里是子 channel 注册到 workerGroup上。

在 register 完成后，子 channel 被挂到了 workerGroup 其中一个线程上，相应的线程如果没有创建也会被创建出来并进入到自己的线程循环中。

当子 channel 注册完毕的时候，childHandler 中 ChannelInitializer#initChannel 会被执行

public final class EchoServer {
 public static void main(String[] args) throws Exception {
  ...
        ServerBootstrap b = new ServerBootstrap();
        b.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);
             }
        });
        ......
 }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.

在 initChannel 把子 channel 的处理类 serverHandler 添加上来了。Netty demo 中对这个处理类的定义非常的简单，仅仅只是打印出来而已。

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    public void channelRead(......) {
        ctx.write(msg);
    }
    ......
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.

五、用户请求到达

当 worker 线程起来以后，会进入线程循环(boss 线程和 worker 线程的 run 函数是一个)。在循环中会遍历自己的任务队列，如果没有任务可处理，便 select 来观察自己所负责的 channel 上是否有事件发生。

public final class NioEventLoop extends SingleThreadEventLoop {

    protected void run() {
        for (;;) {
            if (!hasTasks()) {
                strategy = select(curDeadlineNanos);
            }

            //如果有任务的话就开始处理
            runAllTasks(0);

            //任务处理完毕就调用 epoll_wait 等待事件发生
            processSelectedKeys();
        }
    }
    private int select(long deadlineNanos) throws IOException {
        selector.selectNow();
        ......    
    }
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.

worker 线程会调用 select 发现自己所管理的所有子 channel 上的可读可写事件。在发现有可读事件后，会调用 processSelectedKeys，最后触发 pipeline 使得 EchoServerHandler 方法开始执行。

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    public void channelRead(......) {
        ctx.write(msg);
    }
    ......
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.

六、总结

事实上，Netty 对网络封装的比较灵活。既支持单线程 Reactor，也支持多线程 Reactor、还支持主从多线程 Reactor。三种模型对应的用法如下：

public static void main(String[] args) throws Exception {
 //单线程 Reactor
 EventLoopGroup eventGroup = new NioEventLoopGroup(1);
 ServerBootstrap serverBootstrap = new ServerBootstrap(); 
 serverBootstrap.group(eventGroup);    
 ......

 //多线程 Reactor
 EventLoopGroup eventGroup = new NioEventLoopGroup();
 ServerBootstrap serverBootstrap = new ServerBootstrap(); 
 serverBootstrap.group(eventGroup);    
 ......

 //主从多线程 Reactor
 EventLoopGroup bossGroup = new NioEventLoopGroup(1); 
 EventLoopGroup workerGroup = new NioEventLoopGroup();
 ServerBootstrap serverBootstrap = new ServerBootstrap(); 
 serverBootstrap.group(bossGroup, workerGroup);    
 ......
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.

为了表述的更全面，本文飞哥选择的是最为经典的 主从多线程 Reactor 模式。本文中所描述的内容可以用下面一幅图来表示。

圖片

在 Netty 中的 boss 线程中负责对父 channel（listen socket）上事件的监听和处理，当有新连接到达的时候，选择一个 worker 线程把这个子 channel（连接 socket ）叫给 worker 线程来处理。

其中 Worker 线程就是等待其管理的所有子 channel（连接 socket）上的事件的监听和处理。当发现有事件发生的时候，回调用户设置的 handler 进行处理。在本文的例子中，这个用户 handler 就是 EchoServerHandler#channelRead。

至此，Netty 网络模块的工作核心原理咱们就介绍完了。飞哥一直“鼓吹”内功的好处。只要你具备了坚实的内功，各种语言里看似风牛马不相及的东西，在底层实际上原理是想通的。我本人从来没用 Java 开发过服务器程序，更没碰过 Netty。但是当你多epoll有了深入理解的时候，再看Netty也能很容易看懂，很快就能理解它的核心。这就是锻炼内功的好处！