Netty源码阅读(1)之——客户端源码梗概

目录

准备

开始

NioSocketChannel 的初始化过程

指定

初始化

关于unsafe属性：

关于pipeline的初始化

小结

EventLoopGroup初始化

小结

channel的注册过程

handler的注册过程

 客户端连接

 总结

---

准备

• 源码阅读基于4.1.84.Final版本。
• 从github下载netty项目，并且使用[netty-example]模块
• 你需要先对netty有个大概的了解，比如知道它的模型

开始

找到[netty-example]模块的ceho包，查看简单的使用案例。

public final class EchoClient {
 
    static final String HOST = System.getProperty("host", "127.0.0.1");
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));
 
    public static void main(String[] args) throws Exception {
 
        // Configure the client.
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.TCP_NODELAY, true)
             .handler(new ChannelInitializer() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     //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();
        }
    }
}

• group方法：处理要创建的所有事件
• channel方法：创建channel实例，因为是TCP客户端, 因此使用了 NioSocketChannel
• option方法：配置channel实例
• handler方法：设置数据处理器

上边代码很简单，无非就是新建了个bootstrap，对其进行配置，然后启动客户端连接服务端，最后等待连接关闭。不得不说，netty代码封装的很好，使我们初学者可以简单快速的上手，但是代码封装的越好。意味着内部代码越复杂。

下边，我们来更详细的看下配置bootstrap的流程

NioSocketChannel 的初始化过程

在 Netty 中, Channel 是一个 Socket 的抽象, 它为用户提供了关于 Socket 状态(是否是连接还是断开) 以及对 Socket 的读写等操作. 每当 Netty 建立了一个连接后, 都会有一个对应的 Channel 实例. 

这里，我们需要先知道NioSocketChannel的作用：异步的客户端 TCP Socket 连接。除此之外，还有以下类型：

• NioSocketChannel, 代表异步的客户端 TCP Socket 连接.
• NioServerSocketChannel, 异步的服务器端 TCP Socket 连接.
• NioDatagramChannel, 异步的 UDP 连接
• NioSctpChannel, 异步的客户端 Sctp 连接.
• NioSctpServerChannel, 异步的 Sctp 服务器端连接.
• OioSocketChannel, 同步的客户端 TCP Socket 连接.
• OioServerSocketChannel, 同步的服务器端 TCP Socket 连接.
• OioDatagramChannel, 同步的 UDP 连接
• OioSctpChannel, 同步的 Sctp 服务器端连接.
• OioSctpServerChannel, 同步的客户端 TCP Socket 连接.

指定

从上边代码中我们看到了是channel方法中配置的：

…… 
.channel(NioSocketChannel.class)

 我们点进去查看

public B channel(Class channelClass) {
        return channelFactory(new ReflectiveChannelFactory(
                ObjectUtil.checkNotNull(channelClass, "channelClass")
        ));
    }

发现就是简单的指定了AbstractBootstrap抽象类中的channelFactory属性&&指定了ReflectiveChannelFactory中的constructor属性。同时我们发现实例化ReflectiveChannelFactory中的constructor，也就是实例化public io.netty.channel.socket.nio.NioSocketChannel()是在ReflectiveChannelFactory类中重写的newChannel方法中。那么具体是在哪里调用的呢？这里先挖个坑Ⅰ。

初始化

进入NioSocketChannel类，找到无参构造器，调用有参构造器

public NioSocketChannel(SelectorProvider provider) {
        this(newSocket(provider));
    }

 注意这里的newSocket，是用来打开一个新的 Java NIO SocketChannel。

构造方法继续往下看点点点……来到AbstractChannel

protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }

关于unsafe属性：

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

 我们直接看NioSocketChannelUnsafe的继承和实现，最后可以看出来实现了Unsafe接口，我们来看下这个接口

 一看便知，这些操作都是和 Java 底层的 Socket 相关的操作。

关于pipeline的初始化

首先我们要知道

 Each channel has its own pipeline and it is created automatically when a new channel is created

在实例化一个 Channel 时, 必然伴随着实例化一个 ChannelPipeline.

接下来我们来看一下初始化pipeline时都做了那些事情吧，对着newChannelPipeline()往下一直点点到DefaultChannelPipeline的有参构造。

protected DefaultChannelPipeline(Channel channel) {
        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;
    }

HeadContext的继承结构图，不难发现有ChannelOutboundHandler和ChannelInboundHandler

 TailContext的继承结构图，只有ChannelInboundHandler

 他们又都继承了AbstractChannelHandlerContext，这就是DefaultChannelPipeline中维护的双向链表，tail为尾部，head为头部。这个链表是 Netty 实现 Pipeline 机制的关键

小结

 NioSocketChannel初始化时

• 打开一个新的 Java NIO SocketChannel
• unsafe 通过newUnsafe() 实例化一个 unsafe 对象, 它的类型是 AbstractNioByteChannel.NioByteUnsafe 内部类
• 创建pipeline实例

• readInterestOp变为SelectionKey.OP_READ

• SelectableChannel ch 被配置为非阻塞的 ch.configureBlocking(false)
• SocketChannelConfig config = new NioSocketChannelConfig(this, socket.socket())

EventLoopGroup初始化

点入new NioEventLoopGroup()默认this(0)参数为0。继续走到MultithreadEventLoopGroup的构造方法

super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);

可以看出来，如果new NioEventLoopGroup(？)不填参数，默认就是以下规则。

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

NettyRuntime.availableProcessors() * 2 = 处理器核心数 * 2

最后走到MultithreadEventExecutorGroup的MultithreadEventExecutorGroup方法。

小结

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        checkPositive(nThreads, "nThreads");
		//如果没有自定义执行器(该执行器最终被赋值给EventExecutor的成员变量)，则使用ThreadPerTaskExecutor
        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }
		//实例化children
        children = new EventExecutor[nThreads];
		//for循环将实例化children中的每一个元素
        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
            	//通过子类中的newChild()来实现
                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;
                        }
                    }
                }
            }
        }
		//实例化事件轮询器,即上述的默认的执行器选择工厂DefaultEventExecutorChooserFactory.INSTANCE
        chooser = chooserFactory.newChooser(children);
		//定义异步事件通知，该通知将被添加到事件执行器EventExecutor上，
		//其逻辑也是简单的当children的最后一个元素被成功初始化后设置当前Group的实例化结果
        final FutureListener