Netty源码分析NioEventLoop处理IO事件相关逻辑

目录
  • NioEventLoop的run()方法:
  • processSelectedKeys()方法
  • processSelectedKeysOptimized(selectedKeys.flip())方法
  • processSelectedKey(k, (AbstractNioChannel) a)方法

之前我们了解了执行select()操作的相关逻辑, 这一小节我们继续学习轮询到io事件的相关逻辑:

NioEventLoop的run()方法:

protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    //轮询io事件(1)
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
            }
            cancelledKeys = 0;
            needsToSelectAgain = false;
            //默认是50
            final int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                try {
                    processSelectedKeys();
                } finally {
                    runAllTasks();
                }
            } else {
                //记录下开始时间
                final long ioStartTime = System.nanoTime();
                try {
                    //处理轮询到的key(2)
                    processSelectedKeys();
                } finally {
                    //计算耗时
                    final long ioTime = System.nanoTime() - ioStartTime;
                    //执行task(3)
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        //代码省略
    }
}

我们首先看 if (ioRatio == 100) 这个判断, ioRatio主要是用来控制processSelectedKeys()方法执行时间和任务队列执行时间的比例, 其中ioRatio默认是50, 所以会走到下一步else

首先通过 final long ioStartTime = System.nanoTime() 记录下开始时间, 再通过processSelectedKeys()方法处理轮询到的key

processSelectedKeys()方法

private void processSelectedKeys() {
    if (selectedKeys != null) {
        //flip()方法会直接返回key的数组
        processSelectedKeysOptimized(selectedKeys.flip());
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

我们知道selector通过netty优化之后, 会初始化 selectedKeys这个属性, 所以这个属性不为空就会走到 processSelectedKeysOptimized(selectedKeys.flip()) 方法, 这个方法就是对应优化过的selector进行操作的

如果是非优化的selector, 则会进入 processSelectedKeysPlain(selector.selectedKeys()) 方法

selectedKeys.flip()为selectedKey中绑定的数组, 我们之前小节讲过selectedKeys其实是通过数组存储的, 所以经过select()操作如果监听到事件selectedKeys的数组就会有值

processSelectedKeysOptimized(selectedKeys.flip())方法

private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
    //通过for循环遍历数组
    for (int i = 0;; i ++) {
        //拿到当前的selectionKey
        final SelectionKey k = selectedKeys[i];
        if (k == null) {
            break;
        }
        //将当前引用设置为null
        selectedKeys[i] = null;
        //获取channel(NioSeverSocketChannel)
        final Object a = k.attachment();
        //如果是AbstractNioChannel, 则调用processSelectedKey()方法处理io事件
        if (a instanceof AbstractNioChannel) {
            processSelectedKey(k, (AbstractNioChannel) a);
        } else {
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task);
        }

        //代码省略
    }
}

首先通过for循环遍历数组中的每一个key, 获得key之后首先将数组中对应的下标清空, 因为selector不会自动清空, 这与我们使用原生selector时候, 通过遍历selector.selectedKeys()的set的时候, 拿到key之后要执行remove()是一个意思

之后获取注册在key上的channel, 判断channel是不是AbstractNioChannel, 通常情况都是AbstractNioChannel, 所以这里会执行 processSelectedKey(k, (AbstractNioChannel) a)

processSelectedKey(k, (AbstractNioChannel) a)方法

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    //获取到channel中的unsafe
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    //如果这个key不是合法的, 说明这个channel可能有问题
    if (!k.isValid()) {
        //代码省略
    }
    try {
        //如果是合法的, 拿到key的io事件
        int readyOps = k.readyOps();
        //链接事件
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);
            unsafe.finishConnect();
        }
        //写事件
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            ch.unsafe().forceFlush();
        }
        //读事件和接受链接事件
        //如果当前NioEventLoop是work线程的话, 这里就是op_read事件
        //如果是当前NioEventLoop是boss线程的话, 这里就是op_accept事件
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
            if (!ch.isOpen()) {
                return;
            }
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

我们首先获取和channel绑定的unsafe, 之后拿到channel注册的事件

我们关注

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0)

这个判断, 这个判断相信注释上写的很明白, 如果当前NioEventLoop是work线程的话, 这里就是op_read事件, 如果是当前NioEventLoop是boss线程的话, 这里就是op_accept事件

然后会通过channel绑定的unsafe对象执行read()方法用于处理链接或者读写事件

以上就是NioEventLoop对io事件的处理过程, 有关read()方法执行逻辑, 会在以后的章节中详细剖析,更多关于Netty NioEventLoop处理IO事件逻辑的资料请关注我们其它相关文章!

(0)

相关推荐

  • Netty源码分析NioEventLoop处理IO事件相关逻辑

    目录 NioEventLoop的run()方法: processSelectedKeys()方法 processSelectedKeysOptimized(selectedKeys.flip())方法 processSelectedKey(k, (AbstractNioChannel) a)方法 之前我们了解了执行select()操作的相关逻辑, 这一小节我们继续学习轮询到io事件的相关逻辑: NioEventLoop的run()方法: protected void run() { for (;

  • Netty源码分析NioEventLoop执行select操作入口

    分析完了selector的创建和优化的过程, 这一小节分析select相关操作 select操作的入口,NioEventLoop的run方法: protected void run() { for (;;) { try { switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) { case SelectStrategy.CONTINUE: continue; case SelectStrategy.SEL

  • Netty源码分析NioEventLoop线程的启动

    目录 之前的小节我们学习了NioEventLoop的创建以及线程分配器的初始化, 那么NioEventLoop是如何开启的呢, 我们这一小节继续学习 NioEventLoop的开启方法在其父类SingleThreadEventExecutor中的execute(Runnable task)方法中, 我们跟到这个方法: @Override public void execute(Runnable task) { if (task == null) { throw new NullPointerEx

  • Netty源码分析NioEventLoop初始化线程选择器创建

    前文传送门:NioEventLoop创建 初始化线程选择器 回到上一小节的MultithreadEventExecutorGroup类的构造方法: protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { //代码省略 if (executor == null) { //创建一个新的线程

  • Netty源码解析NioEventLoop创建的构造方法

    目录 前文传送门:Netty源码分析 NioEventLoop 回到上一小节的MultithreadEventExecutorGroup类的构造方法: protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { //代码省略 if (executor == null) { //创建一个新的

  • Netty分布式源码分析监听读事件

    前文传送门:NioSocketChannel注册到selector 我们回到AbstractUnsafe的register0()方法: private void register0(ChannelPromise promise) { try { //省略代码 //做实际的注册 doRegister(); neverRegistered = false; registered = true; //触发事件 pipeline.invokeHandlerAddedIfNeeded(); safeSet

  • 分布式Netty源码分析EventLoopGroup及介绍

    目录 EventLoopGroup介绍 功能1:先来看看注册Channel 功能2:执行一些Runnable任务 EventLoop介绍 NioEventLoop介绍 EpollEventLoop介绍 后续 EventLoopGroup介绍 在前面一篇文章中提到了,EventLoopGroup主要负责2个事情,这里再重复下: 它主要包含2个方面的功能,注册Channel和执行一些Runnable任务. 功能1:先来看看注册Channel 即将Channel注册到Selector上,由Select

  • 分布式Netty源码分析概览

    目录 服务器端demo EventLoopGroup介绍 功能1:先来看看注册Channel 功能2:执行一些Runnable任务 ChannelPipeline介绍 bind过程 sync介绍 误区 4 后续 服务器端demo 看下一个简单的Netty服务器端的例子 public static void main(String[] args){ EventLoopGroup bossGroup=new NioEventLoopGroup(1); EventLoopGroup workerGro

  • Netty分布式Server启动流程服务端初始化源码分析

    目录 第一节:服务端初始化 group方法 初始化成员变量 初始化客户端Handler 第一节:服务端初始化 首先看下在我们用户代码中netty的使用最简单的一个demo: //创建boss和worker线程(1) EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //创建ServerBootstrap(2) ServerBootst

  • Netty分布式NioEventLoop任务队列执行源码分析

    目录 执行任务队列 跟进runAllTasks方法: 我们跟进fetchFromScheduledTaskQueue()方法 回到runAllTasks(long timeoutNanos)方法中 章节小结 前文传送门:NioEventLoop处理IO事件 执行任务队列 继续回到NioEventLoop的run()方法: protected void run() { for (;;) { try { switch (selectStrategy.calculateStrategy(selectN

随机推荐