Netty学习笔记 - 3 (带源码分析部分)

2021年12月

北京

xxd

说明：

源码需要剖析到 Netty 调用 doBind 方法， 追踪到 NioServerSocketChannel 的 doBind并且要 Debug 程序到 NioEventLoop 类 的 run 代码 ，无限循环，在服务器端运行。

//服务器启动类源码

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;


public final class EchoServer {
    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    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();
        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() {
                        @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(new EchoServerHandler());
                        }
                    });
                // 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();
        }
    }
}

解析

先看启动类：main 方法中，首先创建了关于 SSL 的配置类。重点分析下 创建了两个 EventLoopGroup 对象：
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

(1) 这两个对象是整个 Netty 的核心对象，可以说，整个 Netty 的运作都依赖于他们。bossGroup 用于接受Tcp 请求，他会将请求交给 workerGroup ，workerGroup 会获取到真正的连接，然后和连接进行通信，比如读写解码编码等操作。

(2) EventLoopGroup 是 事件循环组（线程组） 含有多个 EventLoop， 可以注册 channel ,用于在事件循环中去进行选择（和选择器相关）…

(3) new NioEventLoopGroup(1); 这个 1 表示 bossGroup 事件组有 1 个线程你可以指定，如果 new NioEventLoopGroup() 会含有默认个线程 cpu 核数*2, 即可以充分的利用多核的优势，

DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

会创建 EventExecutor 数组 children = new EventExecutor[nThreads];
每个元素的类型就是 NIOEventLoop, NIOEventLoop 实现了 EventLoop 接口 和 Executor 接口
try 块中创建了一个 ServerBootstrap 对象，他是一个引导类，用于启动服务器和引导整个程序的初始化（看下源
码 allows easy bootstrap of {@link ServerChannel} ）。它和 ServerChannel 关联， 而 ServerChannel 继承了 Channel，有一些方法 remoteAddress 等
随后，变量 b 调用了 group 方法将两个 group 放入了自己的字段中，用于后期引导使用

(4) 然后添加了一个 channel，其中参数一个 Class 对象，引导类将通过这个 Class 对象反射创建
ChannelFactory。然后添加了一些 TCP 的参数。

**说明：**Channel 的创建在 bind 方法，可以 Debug 下 bind ,会找到 channel = channelFactory.newChannel();
(5) 再添加了一个服务器专属的日志处理器 handler。

(6) 再添加一个 SocketChannel（不是 ServerSocketChannel）的 handler。

(7) 然后绑定端口并阻塞至连接成功。

(8) 最后 main 线程阻塞等待关闭。

(9) finally 块中的代码将在服务器关闭时优雅关闭所有资源

//服务器端处理器源码

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.write(msg);
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised. cause.printStackTrace();
        ctx.close();
    }
}

说明:
这是一个普通的处理器类，用于处理客户端发送来的消息，在我们这里，我们简单的解析出客户端传过
来的内容，然后打印，最后发送字符串给客户端。

构造器方法

public NioEventLoopGroup(int nThreads){
        	this(nThreads,(Executor)null)
        }

上面的 this(nThreads,(Executor)null);调用构造器(通过 alt+d 看即可)

public NioEventLoopGroup(int nThreads,Executor executor){
        	this(nThreads,executor,SelectorProvider.provider());
        }

上面的 this(nThreads,executor,SelectorProvider.provider());调用下面构造器

public NioEventLoopGroup(
        int nThreads,Executor executor,final SelectorProvider selectorProvider){
        this(nThreads,executor,selectorProvider,DefaultSelectStrategyFactory.INSTANCE);
        }

上面的 this()…调用构造器(alt+d)

public NioEventLoopGroup(int nThreads,Executor executor,final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory){
        super(nThreads,executor,selectorProvider,selectStrategyFactory,RejectedExecutionHandlers.reject());
        }

上面的 super()…的方法 是父类： MultithreadEventLoopGroup

protected MultithreadEventLoopGroup(int nThreads,Executor executor,Object...args){
        super(nThreads==0?DEFAULT_EVENT_LOOP_THREADS:nThreads,executor,args);
        }

MultithreadEventExecutorGroup 才 是NioEventLoopGroup 真正的构造方法， 这里可以看成是一个模板方法，使用了设计模式的模板模式(可看我录制视频),所以，我们就需要好好分析 MultithreadEventExecutorGroup 方法

参数说明：
@param nThreads 使用的线程数，默认为 core*2[可以追踪源码]
@param executor 执行器:如果传入 null,则采用 Netty 默认的线程工厂和默认的执行器 ThreadPerTaskExecut

 @param  chooserFactory //单例 new  DefaultEventExecutorChooserFactory()
    @param args //args 在创建执行器的时候传入固定参数
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
        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++) {
            boolean success = false;
            try {
                // 创建 new NioEventLoop
                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 {
                // 如果创建失败，优雅关闭
                if (!success)

                    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);
    final FutureListener