首页 > 编程语言 >Netty源码-01-NioEventLoopGroup

Netty源码-01-NioEventLoopGroup

时间：2022-11-16 22:24:10浏览次数：69  
标签：NioEventLoop Netty nThreads NioEventLoopGroup 源码 线程 executor public

一 定义

摘自源码JavaDoc

/**
 * The {@link EventExecutorGroup} is responsible for providing the {@link EventExecutor}'s to use
 * via its {@link #next()} method. Besides this, it is also responsible for handling their
 * life-cycle and allows shutting them down in a global fashion.
 *
 */
  • EventExecutor是NioEventLoop的抽象
  • NioEventLoopGroup是NioEventLoop管理器
    • 管理NioEventLoop事件循环器
    • 通过next()方法选择一个NioEventLoop事件循环器使用(NioEventLoop线程)

二 类图

截屏2022-11-07 上午10.52.09

从继承关系上可以看出NioEventLoopGroup具有任务管理功能，类似线程池的一种实现。

三 源码

1 事件循环器选择

1.1 Demo

package io.netty.example.basic.eventloopgroup;

import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

/**
 *
 * @since 2022/11/7
 * @author dingrui
 */
public class NioEventLoopGroupTest00 {

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup(3);
        for (int i = 0; i < 4; i++) {
            EventLoop el = group.next();
            System.out.println(el);
        }
    }
}

执行结果为

io.netty.channel.nio.NioEventLoop@2ac1fdc4
io.netty.channel.nio.NioEventLoop@5f150435
io.netty.channel.nio.NioEventLoop@1c53fd30
io.netty.channel.nio.NioEventLoop@2ac1fdc4

可以猜测出来两点

  • NioEventLoopGroup构造方法形参可以控制定义EventLoop数量
  • next()方法从EventLoop集合中选择一个
  • next()方法选择策略具有一定负载均衡功能，大概率是通过定义计数器进行集合取模

1.2 构造方法

// NioEventLoopGroup.java
public NioEventLoopGroup(int nThreads) {
    this(nThreads, (Executor) null);
}

public NioEventLoopGroup(int nThreads, Executor executor) {
        /**
         * executor用于开启NioEventLoop线程所需要的线程执行器
         * SelectorProvider.provider()用于创建selector
         */
        this(nThreads, executor, SelectorProvider.provider());
    }

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

public NioEventLoopGroup(int nThreads,
                             Executor executor, // null
                             final SelectorProvider selectorProvider, // 创建Java的NIO复用器
                             final SelectStrategyFactory selectStrategyFactory
    ) {
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }

// MultithreadEventLoopGroup.java
protected MultithreadEventLoopGroup(int nThreads,
                                        Executor executor, // null
                                        Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

// MultithreadEventExecutorGroup.java
protected MultithreadEventExecutorGroup(int nThreads,
                                            Executor executor, // null
                                            Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }

protected MultithreadEventExecutorGroup(int nThreads, // 标识着group中有几个EventLoop
                                            Executor executor, // null
                                            EventExecutorChooserFactory chooserFactory, // DefaultEventExecutorChooserFactory.INSTANCE
                                            Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        if (executor == null)
            executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); // 构造一个executor线程执行器 一个任务对应一个线程(线程:任务=1:n)

        /**
         * 构建NioEventLoop
         * NioEventLoop children数组 线程池中的线程数组
         */
        this.children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) { // 实例化children数组中的每一个元素
            boolean success = false;
            try {
                /**
                 * children是EventExecutor数组 也就是NioEventLoop集合
                 * {@link MultithreadEventExecutorGroup#newChild(Executor, Object...)}方法是抽象方法 具体实现子类关注
                 * 因为是NioEventLoopGroup调用的该方法 所以最终实现在NioEventLoopGroup中
                 *
                 * 将每个EventLoop和每个线程关联起来
                 */
                children[i] = this.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 ++) { // 但凡有一个child实例化失败 就把已经成功实例化的线程进行shutdown shutdown是异步操作
                        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;
                        }
                    }
                }
            }
        }

        /**
         * 创建线程选择器
         * 线程选择策略
         * NioEventLoop都绑定一个chooser对象 作为线程选择器 通过这个线程选择器 为每一个channel分配不同的线程
         */
        this.chooser = chooserFactory.newChooser(children);

        final FutureListener<Object> terminationListener = new FutureListener<Object>() { // 设置一个listener用来监听线程池中的termination事件 给线程池中的每一个线程都设置这个listener 当监听到所有线程都terminate以后 这个线程池就算真正的terminate了
            @Override
            public void operationComplete(Future<Object> future) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length)
                    terminationFuture.setSuccess(null);
            }
        };

        for (EventExecutor e: children)
            e.terminationFuture().addListener(terminationListener);

        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet); // 只读集合
    }

其中比较重要的几个点

1.2.1 线程执行器Executor
// MultithreadEventExecutorGroup.java
if (executor == null) // 线程执行器
            executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); // 构造一个executor线程执行器 一个任务对应一个线程(线程:任务=1:n)

ThreadFactory仅仅是线程创建的顶层抽象，提供一些必要线程属性

  • 非守护线程(main线程结束任务线程可以继续执行)
  • 线程优先级
// ThreadPerTaskExecutor
public final class ThreadPerTaskExecutor implements Executor { // 只有一个execute方法 每来一个任务就新建一个线程 这个线程池不是给NioEventLoopGroup使用的 而是给NioEventLoop使用的

    /**
     * 负责创建线程
     */
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }

    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start(); // 为每个任务新建一个线程 这个execute方法就是用来开启NioEventLoop线程用的
    }
}

也就是将来在某个时机出发executor的execute()方法，为任务创建线程并执行。

此处的线程并非直接使用的Java实现，而是在此基础上做了一些优化。

1.2.2 事件循环器
// MultithreadEventExecutorGroup.java
/**
         * 构建NioEventLoop
         * NioEventLoop children数组 线程池中的线程数组
         */
        this.children = new EventExecutor[nThreads];

for (int i = 0; i < nThreads; i ++) { // 根据NioEventLoopGroup构造器指定的数量创建NioEventLoop 也就是指定数量的线程数(线程的创建动作延迟到任务提交时)
                /**
                 * 初始化NioEventLoop事件循环器集合 也就是多个线程
                 */
                children[i] = this.newChild(executor, args);
            }
1.2.3 线程选择器
// MultithreadEventExecutorGroup.java
/**
         * 创建线程选择器
         * 线程选择策略
         * NioEventLoopGroup都绑定一个chooser对象 作为线程选择器 通过这个线程选择器 为每一个channel发生的读写IO分配不同的线程进行处理
         */
        this.chooser = chooserFactory.newChooser(children);

netty中提供了2种实现

  • PowerOfTwoEventExecutorChooser
  • GenericEventExecutorChooser

1.3 next

// DefaultEventExecutorChooserFactory.java
private final EventExecutor[] executors; // 在EventLoopGroup构造器中初始化的EventLoop数组

// DefaultEventExecutorChooserFactory.java::PowerOfTwoEventExecutorChooser
/**
         * next()方法的实现就是选择下一个线程的方法
         * 如果线程数是2的倍数 通过位运算 效率高
         */
        @Override
        public EventExecutor next() { // 线程池线程数是2的幂次方 位运算
            return this.executors[idx.getAndIncrement() & this.executors.length - 1];
        }

// DefaultEventExecutorChooserFactory.java::GenericEventExecutorChooser
/*
         * 线程数不是2的倍数 采用绝对值取模的方式 效率一般
         */
        @Override
        public EventExecutor next() { // 线程池线程数量不是2的幂次方 采用取模方式
            return this.executors[(int) Math.abs(idx.getAndIncrement() % this.executors.length)];
        }

原理比较简单，本质就是对数组长度取模，在一定场景下用位运算替代取模运算。

也就EventLoopGroup这个管理器按照轮询方式从NioEventLoop数组中选择事件循环器执行操作，起到了一定的负载效果。

2 (定时)任务提交

2.1 Demo

// Demo.java
package io.netty.example.basic.eventloopgroup;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

import java.util.concurrent.TimeUnit;

/**
 *
 * @since 2022/11/7
 * @author dingrui
 */
public class NioEventLoopGroupTest01 {

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        group.execute(() -> {
            System.out.println("execute task...");
        });
        group.submit(() -> {
            System.out.println("submit task...");
        });
        group.schedule(() -> {
                    System.out.println("schedule task...");
                },
                5_000,
                TimeUnit.MILLISECONDS);
        System.out.println("main thread end");
    }
}

执行结果

main thread end
submit task...
execute task...

上面已经分析过

  • EventLoopGroup仅仅是一个管理器，本身不具备执行任务能力
  • 管理器暴露next()选择其内部真正的执行单元(事件循环器)进行工作
  • 每个EventLoop都是一个线程，真正处理任务的执行单元，非守护线程
  • NioEventLoop是特定处理NIO多路复用这种场景的具体实现，因此还通过组合了Java中Selector多路复用器实现基于事件的响应模式React

2.2 任务执行单元

下面这几个方法都是EventLoopGroup管理器通过选择器chooser选择真正的执行单元EventLoop事件循环器来处理的

  • execute(...)
  • submit(...)
  • schedule(...)
  • scheduleAtFixedRate(...)
  • scheduleWithFixedDelay(...)

以execute(...)方法为例

// Demo.java
group.execute(() -> {
            System.out.println("execute task...");
        });

// AbstractEventExecutorGroup.java
public void execute(Runnable command) {
        this.next().execute(command);
    }

// MultithreadEventLoopGroup.java
@Override
    public EventLoop next() { // 从线程池NioEventLoopGroup中选择一个线程NioEventLoop
        return (EventLoop) super.next();
    }

// MultithreadEventExecutorGroup
@Override
    public EventExecutor next() {
        return this.chooser.next();
    }

那么真正执行execute(...)方法的就是EventLoop。

标签：NioEventLoop,Netty,nThreads,NioEventLoopGroup,源码,线程,executor,public
From： https://www.cnblogs.com/miss-u/p/16897723.html

相关文章

  • Netty源码-02-FastThreadLocalThread
    一DemopublicclassFastThreadLocalTest00{privatefinalstaticFastThreadLocal<Long>v=newFastThreadLocal<Long>(){@Overrideprotec......
  • apache启动遇到phpinfo只显示源码问题
    在安装php和apache的时候，会遇到只显示源码的问题网上找了好多帖子都是在改php.ini的东西，但是改了半天也不对，发现我安装的wordpress目录也打不开，所以我认为这就是apache服......
  • python源码通过词语标记化器tokenize提取注释并正则匹配测试用例作者名
    提取代码如下importtokenizeimportrewithtokenize.open('readcomment.py')asf:list=[]fortoktype,tok,start,end,lineintokenize.generate_t......
  • SpringBoot源码解析
    1.创建SpringApplication对象1.1webApplicationType设置（SERVLET）1.2setInitializers设置通过getSpringFactoriesInstances方法从META-INF/spring.factories下获取k......
  • (笔者推荐)【Java权威指南】「官方文档-中英互译」AQS的源码注释分析，贯穿总体核心流程
    前提说明本文主要针对于Java官方文档中的先关的官方注释进行中英文互译，保证了源码坐着的设计思路以及相关知识技能介绍分析等，本文主要进行介绍AQS的源码官方注释的含义介绍，......
  • vue源码分析-插槽原理
    Vue组件的另一个重要概念是插槽，它允许你以一种不同于严格的父子关系的方式组合组件。插槽为你提供了一个将内容放置到新位置或使组件更通用的出口。这一节将围绕官网对插......
  • Spring源码整体脉络
    BeanFactory:Spring顶层核心接口，使用了简单工厂模式，负责生产Bean。BeanDefinition：Spring顶层核心接口，封装了生产Bean的一切原料。从读取配置到扫描到注册bean，主要用到......
  • @PreDestroy与@PostConstruct关键源码实现
    CommonAnnotationBeanPostProcessor(继承InitDestroyAnnotationBeanPostProcessor)在InitDestroyAnnotationBeanPostProcessor中，buildLifecycleMetadata方法会将目标类中......
  • MySQL 源码解读之-语法解析（三）
    MySQL源码解读之-语法解析（三）在前两篇文章中已经讲述了bison如何解析sql语句并生成AST树。那么MySQL是如何和bison的程序关联起来的呢，并通过gdb调试一下。在MyS......
  • CentOS服务器上普通用户（非root）源码部署禅道
    简介：公司服务器上的docker容器，其中一个容器最小化安装了CentOS，要在这个什么命令都没有的Linux系统上，在指定的路径下，部署禅道（先搭建环境）；搭建Apache、PHP、MariaDB，并结合同......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99