Java进阶-Netty-进阶

一、Reactor线程

??源码基于4.1.6.Final版本。

1.1 Reactor线程启动

??NioEventLoop的run方法是reactor线程的主体，在第一次添加任务的时候被启动。

• 入口：NioEventLoop父类SingleThreadEventExecutor的execute方法

@Override
public void execute(Runnable task) {
    ...
    boolean inEventLoop = inEventLoop();
    if (inEventLoop) {
        addTask(task);
    } else {
        startThread();
        addTask(task);
        ...
    }
    ...
}

• netty的reactor线程在添加一个任务的时候被创建，该线程实体为FastThreadLocalThread。
• 最后线程执行主体为NioEventLoop的run方法。

1.2 Reactor线程执行

@Override
protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
                    // fallthrough
            }
            processSelectedKeys();
            runAllTasks(...);
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        ...
    }

• 轮询注册到reactor线程对应的selector上的所有的channel的IO事件

select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
      selector.wakeup();
}

定时任务截止事时间快到了，中断本次轮询
轮询过程中发现有任务加入，中断本次轮询
阻塞式select操作

??netty会在每次进行selector.select(timeoutMillis)之前记录一下开始时间currentTimeNanos，在select之后记录一下结束时间，判断select操作是否至少持续了timeoutMillis秒，如果持续的时间大于等于timeoutMillis，说明就是一次有效的轮询，重置selectCnt标志，否则，表明该阻塞方法并没有阻塞这么长时间，可能触发了jdk的空轮询bug，当空轮询的次数超过一个阀值的时候，默认是512，就开始重建selector。

• 处理产生网络IO事件的channel

processSelectedKeys()；

private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized(selectedKeys.flip());
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

取出IO事件以及对应的netty channel类
处理该channel
判断是否该再来次轮询

对于boss NioEventLoop来说，轮询到的是基本上就是连接事件，后续的事情就通过他的pipeline将连接扔给一个worker NioEventLoop处理
对于worker NioEventLoop来说，轮询到的基本上都是io读写事件，后续的事情就是通过他的pipeline将读取到的字节流传递给每个channelHandler来处理SelectedSelectionKeySet

??netty使用数组替换掉jdk原生的HashSet来保证IO事件的高效处理，每个SelectionKey上绑定了netty类AbstractChannel对象作为attachment，在处理每个SelectionKey的时候，就可以找到AbstractChannel，然后通过pipeline的方式将处理串行到ChannelHandler，回调到用户方法。

• 处理任务队列

用户自定义普通任务
非当前reactor线程调用channel的各种方法
用户自定义定时任务：1）若干时间后执行一次 2）每隔一段时间执行一次 3）每次执行结束，隔一定时间再执行一次

??taskQueue在NioEventLoop中默认是mpsc队列，mpsc队列，即多生产者单消费者队列，netty使用mpsc，方便的将外部线程的task聚集，在reactor线程内部用单线程来串行执行。

??reactor线程task调度：

• 从scheduledTaskQueue转移到期的定时任务到taskQueue(mpsc queue)
• 计算本次任务循环的截止时间
• 执行任务
• 收尾

当前reactor线程调用当前eventLoop执行任务，直接执行，否则，添加到任务队列稍后执行
netty内部的任务分为普通任务和定时任务，分别落地到MpscQueue和PriorityQueue
netty每次执行任务循环之前，会将已经到期的定时任务从PriorityQueue转移到MpscQueue
netty每隔64个任务检查一下是否该退出任务循环

二、服务端启动

b.bind(8888).sync();

public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
} 

public ChannelFuture bind(SocketAddress localAddress) {
    validate();
    if (localAddress == null) {
        throw new NullPointerException("localAddress");
    }
    return doBind(localAddress);
}

private ChannelFuture doBind(final SocketAddress localAddress) {
    //...
    final ChannelFuture regFuture = initAndRegister();
    //...
    final Channel channel = regFuture.channel();
    //...
    doBind0(regFuture, channel, localAddress, promise);
    //...
    return promise;
}

• new一个channel

用户调用方法Bootstrap.bind(port)第一步就是通过反射的方式new一个NioServerSocketChannel对象，并且在new的过程中创建了一系列的核心组件。

• init这个channel

设置option和attr
设置新接入channel的option和attr
加入新连接处理器

• 将这个channel register到某个对象

1）设置启动类参数，最重要的就是设置channel创建server对应的channel，创建各大组件，包括ChannelConfig,ChannelId,ChannelPipeline,ChannelHandler,Unsafe等
2）初始化server对应的channel，设置一些attr，option，以及设置子channel的attr，option，给server的channel添加新channel接入器，并触发addHandler,register等事件
3）调用jdk底层做端口绑定，并触发active事件，active触发的时候，真正做服务端口绑定

三、新连接接入

所有的channel底层都会有一个与unsafe绑定，每种类型的channel实际的操作都由unsafe来实现

??流水线的开始就是HeadContxt，流水线的结束就是TailConext，HeadContxt中调用Unsafe做具体的操作，TailConext中用于向用户抛出pipeline中未处理异常以及对未处理消息的警告。

• 检测到有新连接进入
• 将新的连接注册到worker线程组
• 注册新连接的读事件

//NioEventLoop.java
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    int readyOps = k.readyOps();
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
        unsafe.read();
    }
}

• boos reactor线程轮询到有新的连接进入
• 通过封装jdk底层的channel创建NioSocketChannel以及一系列的netty核心组件
• 将该条连接通过chooser，选择一条worker reactor线程绑定上去
• 注册读事件，开始新连接的读写

四、pipeline

4.1 pipeline 初始化

??pipeline是channel其中的一员，在AbstractChannel中被创建。

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}
protected DefaultChannelPipeline newChannelPipeline() {
    return new DefaultChannelPipeline(this);
}
protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

??pipeline中保存了channel的引用，默认情况下，一条pipeline会有两个节点，head和tail。

4.2 pipeline添加节点

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         p.addLast(new Spliter())
         p.addLast(new Decoder());
         p.addLast(new BusinessHandler())
         p.addLast(new Encoder());
     }
});

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        // 1.检查是否有重复handler
        checkMultiplicity(handler);
        // 2.创建节点
        newCtx = newContext(group, filterName(name, handler), handler);
        // 3.添加节点
        addLast0(newCtx);
    }
   
    // 4.回调用户方法
    callHandlerAdded0(handler);
    
    return this;
}

netty中用两个字段来表示这个channelHandlerContext属于inBound还是outBound，或者两者都是。

private static boolean isInbound(ChannelHandler handler) {
    return handler instanceof ChannelInboundHandler;
}

private static boolean isOutbound(ChannelHandler handler) {
    return handler instanceof ChannelOutboundHandler;
}

• 检查是否有重复handler
• 创建节点
• 添加节点
• 回调用户方法

private void addLast0(AbstractChannelHandlerContext newCtx) {
    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev; // 1
    newCtx.next = tail; // 2
    prev.next = newCtx; // 3
    tail.prev = newCtx; // 4
}

4.3 pipeline删除节点

@Override
public final ChannelPipeline remove(ChannelHandler handler) {
    remove(getContextOrDie(handler));
    
    return this;
}

• 找到待删除的节点
• 调整双向链表指针删除
• 回调用户函数

4.4 pipeline其他

• 一个Channel对应一个Unsafe，Unsafe处理底层操作，NioServerSocketChannel对应NioMessageUnsafe, NioSocketChannel对应NioByteUnsafe
• inBound事件从head节点传播到tail节点，outBound事件从tail节点传播到head节点
• 异常传播只会往后传播，而且不分inbound还是outbound节点，不像outBound事件一样会往前传播

五、writeAndFlush

• pipeline中的标准链表结构

• Encoder节点分配一个ByteBuf，调用encode方法，将java对象根据自定义协议写入到ByteBuf，然后再把ByteBuf传入到下一个节点。
• pipeline中的编码器原理是创建一个ByteBuf，将java对象转换为ByteBuf，然后再把ByteBuf继续向前传递
• 调用write方法并没有将数据写到Socket缓冲区中，而是写到了一个单向链表的数据结构中，flush才是真正的写出
• writeAndFlush等价于先将数据写到netty的缓冲区，再将netty缓冲区中的数据写到Socket缓冲区中，写的过程与并发编程类似，用自旋锁保证写成功
• netty中的缓冲区中的ByteBuf为DirectByteBuf

六、拆包器

• 不断从TCP缓冲区中读取数据，每次读取完都需要判断是否是一个完整的数据包
• 如果当前读取的数据不足以拼接成一个完整的业务数据包，那就保留该数据，继续从tcp缓冲区中读取，直到得到一个完整的数据包
• 如果当前读到的数据加上已经读取的数据足够拼接成一个数据包，那就将已经读取的数据拼接上本次读取的数据，够成一个完整的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下次读到的数据尝试拼接

netty中的拆包内部会有一个累加器，每次读取到数据都会不断累加，然后尝试对累加到的数据进行拆包，拆成一个完整的业务数据包，这个基类叫做ByteToMessageDecoder

netty将具体如何拆包抽象出一个decode方法，不同的拆包器实现不同的decode方法，就能实现不同协议的拆包