Netty 源计划之端启动流程(一)

╰+攻爆jí腚メ 2023-09-28 13:22 108阅读 0赞

### 前言 ###

Netty 源计划系列是根据自顶向下的方式去解读Netty的源码，从使用的角度去一步一步解析Netty做了什么。  
阅读该系列需要有一定的有关Netty的知识，建议先去看博主的《Netty的基本知识》。

### 环境要求 ###

*  JDK：11+
 *  Netty：4.1.78.Final

### 概述 ###

这一章是 Netty 源计划系列的首章，我打算在这一章中, 展示一下 Netty 的客户端和服务端的初始化和启动的流程, 给读者一个对 Netty 源码有一个大致的框架上的认识。  
本章会从 Bootstrap/ServerBootstrap 类 入手, 分析 Netty 程序的初始化和启动的流程，然后下面是Netty总的一个流程图，可以帮助大家来解读代码。

![在这里插入图片描述][4a7cbd3f1ef54c36bd4e7c44bb43ae2c.png]

### Bootstrap ###

Bootstrap 是 Netty 提供的一个便利的工厂类, 我们可以通过它来完成 Netty 的客户端 初始化。下面我以 Netty 源码例子中的 Echo作为例子, 从客户端分别分析一下Netty 的程序是如何启动的。

首先, 让我们从客户端方面的代码开始，下面是源码**example/src/main/java/io/netty/example/echo/EchoClient.java** 的客户端部分的启动代码:

EventLoopGroup group = new NioEventLoopGroup();
            try {
        
                Bootstrap b = new Bootstrap();
                b.group(group)
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.TCP_NODELAY, true)
                 .handler(new ChannelInitializer<SocketChannel>() {
        
                     @Override
                     public void initChannel(SocketChannel ch) throws Exception {
        
                         ChannelPipeline p = ch.pipeline();
                         if (sslCtx != null) {
        
                             p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
                         }
                         //p.addLast(new LoggingHandler(LogLevel.INFO));
                         p.addLast(new EchoClientHandler());
                     }
                 });
    
                // Start the client.
                ChannelFuture f = b.connect(HOST, PORT).sync();
    
                // Wait until the connection is closed.
                f.channel().closeFuture().sync();
            } finally {
        
                // Shut down the event loop to terminate all threads.
                group.shutdownGracefully();
            }

从上面的客户端代码虽然简单, 但是却展示了 Netty 客户端初始化时所需的所有内容:

1.  group(): 指定实现的Reactor模型。不论是服务器端还是客户端, 都必须指定 EventLoopGroup，在这个例子中, 指定了 NioEventLoopGroup, 表示一个 NIO 的EventLoopGroup。
2.  channel(): 指定 Channel 的类型. 因为是客户端, 因此使用了 NioSocketChannel。
3.  option(): 配置Channel的选项。
4.  handler(): 设置数据的处理器。

下面我们深入代码, 看一下客户端通过 Bootstrap 启动后, 都做了哪些工作。

#### group() ####

我们看一下group方法干了什么：

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

可以看到应该是为Bootstrap创建一个指定的EventLoopGroup，也就是NioEventLoopGroup。一开始的  
**EventLoopGroup group = new NioEventLoopGroup()**;最终是调用了**MultithreadEventExecutorGroup**构造器：

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                                EventExecutorChooserFactory chooserFactory, Object... args) {
        
            //如果传入的线程数小于0则抛出异常
            if (nThreads <= 0) {
        
                throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
            }
            //如果传入的执行器是空的则采用默认的线程工厂和默认的执行器
            if (executor == null) {
        
                executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
            }
            //创建指定线程数的执行器数组
            children = new EventExecutor[nThreads];
            //遍历创建执行器
            for (int i = 0; i < nThreads; i ++) {
        
                //是否创建成功默认是false
                boolean success = false;
                try {
        
                    //使用了newChild方法创建执行器并且传入了executor和设置参数args
                    children[i] = newChild(executor, args);
                    //未报异常则设置true
                    success = true;
                } catch (Exception e) {
        
                    //创建失败则抛出异常
                    throw new IllegalStateException("failed to create a child event loop", e);
                } finally {
        
                    //如果success是false则代表创建失败则将已经创建好的执行器进行关闭
                    if (!success) {
        
                        //此处则是遍历创建了i的执行去调用他的shutdown方法
                        for (int j = 0; j < i; j ++) {
        
                            children[j].shutdownGracefully();
                        }
                        //虽然上面调用了中断方法但是他并不会立马终止，因为内部还有内容需要执行。
                        for (int j = 0; j < i; j ++) {
        
                            EventExecutor e = children[j];
                            try {
        
                                //判断当前的执行器是否终止了如果没有则等待获取结果
                                while (!e.isTerminated()) {
        
                                    e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                                }
                            } catch (InterruptedException interrupted) {
        
                                //抛出异常则中断当前线程
                                // Let the caller handle the interruption.
                                Thread.currentThread().interrupt();
                                break;
                            }
                        }
                    }
                }
            }
            //获取执行器的选择器
            chooser = chooserFactory.newChooser(children);
            //创建一个future的监听器用于监听终止结果
            final FutureListener<Object> terminationListener = new FutureListener<Object>() {
        
                @Override
                public void operationComplete(Future<Object> future) throws Exception {
        
                    //当此执行组中的执行器被关闭的时候回调用此方法进入这里，这里进行终止数加一然后比较是否已经达到了执行器的总数
                    //如果没有则跳过，如果有则设置当前执行器的终止future为success为null
                    if (terminatedChildren.incrementAndGet() == children.length) {
        
                        terminationFuture.setSuccess(null);
                    }
                }
            };
            //遍历创建好的执行器动态添加终止future的结果监听器，当监听器触发则会进入上方的内部类实现
            for (EventExecutor e: children) {
        
                e.terminationFuture().addListener(terminationListener);
            }
            //创建一个children的镜像set
            Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
            //拷贝这个set
            Collections.addAll(childrenSet, children);
            //并且设置此set内的所有数据不允许修改然后返回设置给readonlyChildren
            readonlyChildren = Collections.unmodifiableSet(childrenSet);
        }

也就是这个方法给Bootstrap设置了任务处理器组EventLoopGroup，而这个组可能包含多个任务处理器EventLoop。

#### channel() ####

了解Channel之前，我们先知道Socket是什么，它实质上提供了进程通信的端点，进程通信之前，双方首先必须各自创建一个端点，否则是没有办法建立联系并相互通信的。正如打电话之前，双方必须各自拥有一台电话机一样。有兴趣的可以去看下博主的《unix系统之套接字通信》。  
在 Netty 中, Channel 是一个 Socket 的抽象，也就是说Channel是连接的门。

我们看一下channel方法干了什么：

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

可以看到应该是为Bootstrap创建一个指定Channel类型的ChannelFactory，这样的话利用这个工厂来创建Channel实例，进去**ReflectiveChannelFactory**可以看到存的是Channel的class构造器，那我们就懂了Netty想通过构造器反射来创建Channel实例(这个可以看ReflectiveChannelFactory后面的newChannel方法也证实)

public ReflectiveChannelFactory(Class<? extends T> clazz) {
        
            ObjectUtil.checkNotNull(clazz, "clazz");
            try {
        
                this.constructor = clazz.getConstructor();
            } catch (NoSuchMethodException e) {
        
                throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                        " does not have a public non-arg constructor", e);
            }
        }
    
        public T newChannel() {
        
            try {
        
                return constructor.newInstance();
            } catch (Throwable t) {
        
                throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
            }
        }

#### option() ####

这个方法为Channel服务的，让用户通过构建者模式来自定义它的选项，进去后可以看到是ChannelOption，有哪些选项都可以在ChannelOption里查看，这里便不列举了。

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

#### handler() ####

先进去看看方法：

public B handler(ChannelHandler handler) {
        
            this.handler = ObjectUtil.checkNotNull(handler, "handler");
            return self();
        }

可以看到核心就是设置Bootstrap的handler，那我们看看ChannelHandler是个什么鬼，下面是类的注释：

> Handles an I/O event or intercepts an I/O operation, and forwards it to its next handler in its ChannelPipeline.

总结就是说ChannelHandler是ChannelPipeline里的处理器，而这些处理器是基于责任链模式来实现的。

![在这里插入图片描述][4b676c2998a94585b057067c54209a3d.png]

那ChannelPipeline就是数据流进Channel后，得进入这个通道来处理的东西了。  
那就是说我们可以通过handle方法来定义怎么处理数据。

#### connect() ####

Bootstrap的初始化完成后，我们就可以来连接服务端了，而connect就是这个的入口，下面是源码，可以看到核心就是**doResolveAndConnect**方法了。

public ChannelFuture connect(SocketAddress remoteAddress) {
        
            ObjectUtil.checkNotNull(remoteAddress, "remoteAddress");
            // 校验下Bootstrap的成员变量有没有空的
            validate();
            return doResolveAndConnect(remoteAddress, config.localAddress());
        }

看看**doResolveAndConnect**

private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
        
        // 需要重点分析的方法
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
    	
        if (regFuture.isDone()) {
        
            if (!regFuture.isSuccess()) {
        
                return regFuture;
            }
            // 真正的连接方法
            return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
        } else {
        
            // Registration future is almost always fulfilled already, but just in case it's not.
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            // 没有完全执行完成注册与初始化操作，添加一个事件监听器，此处就实现了Netty中Future-Listener机制
            regFuture.addListener(new ChannelFutureListener() {
        
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
        
                    // Directly obtain the cause and do a null check so we only need one volatile read in case of a
                    // failure.
                    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();
                        doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }

对于\*\* final ChannelFuture regFuture = initAndRegister();\*\*

final ChannelFuture initAndRegister() {
        
        Channel channel = null;
        try {
        
            // 从工厂中创建channel实例
            channel = channelFactory.newChannel();
            // 初始化创建的channel，该方法此前还未进行追溯，在此处重点分析
            init(channel);
            
        } catch (Throwable t) {
        
            // 省略异常处理代码
        }
    	// 该方法需要重点进行追溯，查看NioEventLoopGroup是如何创建ChannelFuture的，注意此处传入了创建后的Channel
        ChannelFuture regFuture = config().group().register(channel);
        
        if (regFuture.cause() != null) {
        
            if (channel.isRegistered()) {
        
                channel.close();
            } else {
        
                channel.unsafe().closeForcibly();
            }
        }
        
        return regFuture;
    }

其中\*\*channelFactory.newChannel()\*\*会通过构造器来创建channel实例，最终指到AbstractChannel，所以就是创建了一个包含Unsafe与ChannelPipeline的channel。

protected AbstractChannel(Channel parent) {
        
        this.parent = parent;
        // 创建channel的id
        id = newId();
        // 创建一个Unsafe对象
        unsafe = newUnsafe();
        // 创建一个ChannelPipeline对象
        pipeline = newChannelPipeline();
    }

Unsafe对象其实是对Java底层Socket操作的封装对象，是沟通Netty和Java的重要桥梁。  
调用的newUnsafe();方法的实际调用为AbstractNioByteChannel中的newUnsafe，可以看到new了一个NioSocketChannelUnsafe，而它是实现了Unsafe接口的一个对象：

@Override
    protected AbstractNioUnsafe newUnsafe() {
        
        return new NioSocketChannelUnsafe();
    }

而\*\*newChannelPipeline()\*\*就好理解了，直接定位到DefaultChannelPipeline：

protected DefaultChannelPipeline(Channel channel) {
        
        // 将创建的channel作为pipline的一个属性进行赋值
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);
    	
        // 管道初始被创建时有头结点和尾结点
        tail = new TailContext(this);
        head = new HeadContext(this);
    
        head.next = tail;
        tail.prev = head;
    }

其实是一个以ChannelHandlerContext为节点的双向链表，主要用来存放handler，一个pipline与一个Channel相关联。

那channel实例化完成后，接下来就是 \*\*init(channel);\*\*了：

void init(Channel channel) {
        
        // 从创建的NioChannel中获取管道，该方法无需追溯
        ChannelPipeline p = channel.pipeline();
        
        // 将链式调用时初始化的handler参数添加进行创建的管道中
        p.addLast(config.handler());
    	
        setChannelOptions(channel, newOptionsArray(), logger);
        setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_》ARRAY));
    }

初始化完channel之后，就要注册它了**ChannelFuture regFuture = config().group().register(channel);**  
最终会执行到**SingleThreadEventLoop\#register**：

public ChannelFuture register(Channel channel) {
        
            return register(new DefaultChannelPromise(channel, this));
    }
    
    public ChannelFuture register(final ChannelPromise promise) {
        
            ObjectUtil.checkNotNull(promise, "promise");
            promise.channel().unsafe().register(this, promise);
            return promise;
        }

注册的过程就是把Eventloop与Channel丢到ChannelPromise里，那ChannelPromise是什么，在回答这个之前，我们知道程序处理完返回的是一个ChannelFuture，这是干嘛的?  
Netty所有的I/O操作都是异步的。因为一个操作可能不会立即返回，所以我们需要一种用于在之后得某个时间点确定其结果的方法。为此，Netty提供了ChannelFuture接口，其addListener()方法注册了一个ChannelFutureListener，以便在某个操作完成时（无论是否成功）得到通知。  
Netty可以说是有Channel、EventLoop、ChannelFuture聚合起来的一个网络抽象代表

*  Channel——Socket；
 *  EventLoop——控制流、多线程处理、并发
 *  ChannelFuture——异步通知

所以很明显这是一个Future与Promise异步编程的典型了，那ChannelPromise就是用来**回调异步执行的结果**，上面提到的监听器的处理也是由它实现。

现在整个initAndRegister()都已经完成！最终要实现真正的连接了，**doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise())**,最终会执行到doConnect0：

private static void doConnect0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress remoteAddress, 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.
        channel.eventLoop().execute(new Runnable() {
        
            @Override
            public void run() {
        
                if (regFuture.isSuccess()) {
        
                    if (localAddress == null) {
        
                        channel.connect(remoteAddress, promise);
                    } else {
        
                        channel.connect(remoteAddress, localAddress, promise);
                    }
                    promise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                } else {
        
                    promise.setFailure(regFuture.cause());
                }
            }
        });
    }

也就是最终会调用到Channel的connect进行连接。  
也就是执行到这里，客户端连接服务器就成功了。

### ServerBootstrap ###

ServerBootstrap是 Netty 提供的一个便利的工厂类, 我们可以通过它来完成 Netty 的服务端 初始化。与Bootstrap一样，都是Echo的例子。

首先, 让我们从服务端方面的代码开始，下面是源码**example/src/main/java/io/netty/example/echo/EchoServer.java** 的服务端部分的启动代码:

public static void main(String[] args) throws Exception {
        
            // Configure SSL.
            final SslContext sslCtx;
            if (SSL) {
        
                SelfSignedCertificate ssc = new SelfSignedCertificate();
                sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
            } else {
        
                sslCtx = null;
            }
    
            // Configure the server.
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            final EchoServerHandler serverHandler = new EchoServerHandler();
            try {
        
                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(new LoggingHandler(LogLevel.INFO));
                         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();
            }
        }

从上面的代码很明显跟客户端的差不多，下面列举出差异性：

1.  group(): 服务器端需要指定两个 EventLoopGroup, 一个是 bossGroup, 用于处理客户端的连接请求; 另一个是 workerGroup, 用于处理与各个客户端的 IO 操作。
2.  handler(): 客户端连接的处理器,跟Bootstarp的handler()一毛一样。
3.  childHandler(): 客户端的IO数据的处理器。

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

跟客户端一样，只不过创建了2个EventLoopGroup而已。

#### childHandler() ####

先进去看看方法：

public ServerBootstrap childHandler(ChannelHandler childHandler) {
        
            this.childHandler = ObjectUtil.checkNotNull(childHandler, "childHandler");
            return this;
        }

那就是跟handle()大同小异了。

#### bind() ####

ServerBootstrap的初始化完成后，我们就可以来绑定端口了，而bind就是这个的入口。  
**ChannelFuture f = b.bind(PORT).sync();**

public ChannelFuture bind(SocketAddress localAddress) {
        
            validate();
            return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
        }

\*\* \*\*核心代码doBing():

private ChannelFuture doBind(final SocketAddress localAddress) {
        
            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(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;
            }
        }

很明显还是老套路先注册跟初始化Channel，最终调到doBind0():

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.
            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的bind进行绑定端口。  
也就是执行到这里，服务器搭建就成功了。

### 总结 ###

从代码角度来看，Netty的客户端与服务端的代码基本一致，通过Socket编程的bind与connect来达到网络交互。代码写的很清晰，看的很爽。

[4a7cbd3f1ef54c36bd4e7c44bb43ae2c.png]: https://img-blog.csdnimg.cn/4a7cbd3f1ef54c36bd4e7c44bb43ae2c.png
[4b676c2998a94585b057067c54209a3d.png]: https://img-blog.csdnimg.cn/4b676c2998a94585b057067c54209a3d.png