Java NIO Selector选择器源码分析

这篇具有很好参考价值的文章主要介绍了Java NIO Selector选择器源码分析。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。


前言

Java NIO(New I/O)的Selector选择器是一个用于多路复用(Multiplexing)的I/O操作的关键组件。它允许一个单独的线程监视多个通道(Channel)的可读性和可写性,从而有效地管理大量并发连接。

Selector类结构

Java NIO Selector选择器源码分析,网络编程,Java,nio,java,Selector

Selector抽象类

public abstract class Selector implements Closeable {
 
    protected Selector() { }
 
    // 创建Selector对象
    public static Selector open() throws IOException {
        return SelectorProvider.provider().openSelector();
    }
 
    // 检测Selector是否打开
    public abstract boolean isOpen();
 
    // 返回创建该Selector的Provider
    public abstract SelectorProvider provider();
 
    // 返回Key集合,key集合不能被直接修改,只有在被cancel和channel被撤销的时候key才被移除。并且不是线程安全的集合。
    public abstract Set<SelectionKey> keys();
 
    // 返回selected-key集合,key可以直接移除,但是不可以直接增加。并且不是线程安全的集合。
    public abstract Set<SelectionKey> selectedKeys();
 
     // 选择channel有IO事件的key。
     // 该方法是非阻塞的selection操作,如果自上次selection操作之后无channel具有IO事件,该方法会立刻返回零。
     // 执行该方法会立刻清除之前执行的wakeup影响。
    public abstract int selectNow() throws IOException;
 
    // 阻塞操作,只有在以下的状态变化时:
    //(1)至少有一个IO的channel(2)调用selector.wakeup方法(3)当前线程被interrupt(4)timeout时间到(毫秒)
    public abstract int select(long timeout)
        throws IOException;
 
     // 阻塞操作,返回条件与select(long timeout)类似
    public abstract int select() throws IOException;
 
     // 唤醒当前select阻塞的操作:如果另一个线程当前阻塞在select或select(long)方法。
     // 如果当前没有select阻塞,则下次执行select或select(long)则直接返回,除非selectNow同时执行;
     //之后select和select(long)方法会正常阻塞;
     // 如果在select操作之间多次调用wakeup与调用一次效果是一样的
    public abstract Selector wakeup();
 
    // 关闭Selector。
    // 调用close方法,如果当前阻塞在selection操作,就像调用wakeup方法一样会立刻中断操作
    // 与该selector关联的未cancelled的key将失效,它们的channel将撤销,与Selector相关的其他资源将释放。
    // 如果Selector已经关闭,执行这个方法将没有影响。
    // selector关闭之后,如果执行与selector相关的操作会报ClosedSelectorException
    public abstract void close() throws IOException;
 
}
// java.nio.channels.spi.SelectorProvider
    public static SelectorProvider provider() {
    synchronized (lock) {
        if (provider != null)
            return provider;
        return AccessController.doPrivileged(
            new PrivilegedAction<SelectorProvider>() {
                public SelectorProvider run() {
                        if (loadProviderFromProperty())
                            return provider;
                        if (loadProviderAsService())
                            return provider;
                            // 这里就是打开Selector的真正方法
                        provider = sun.nio.ch.DefaultSelectorProvider.create();
                        return provider;
                    }
                });
    }
}

AbstractSelector

AbstractSelector主要实现了Selector的打开关闭的状态维护,支持异步关闭和中断的begin和end方法,cancelledKeys等。

public abstract class AbstractSelector
    extends Selector
{
    private AtomicBoolean selectorOpen = new AtomicBoolean(true); // 是否打开
    // The provider that created this selector
    private final SelectorProvider provider;
    protected AbstractSelector(SelectorProvider provider) {
        this.provider = provider;
    }
    // 三大key集合之一cancelledKeys
    private final Set<SelectionKey> cancelledKeys = new HashSet<SelectionKey>();
 
    void cancel(SelectionKey k) {                       // package-private
        synchronized (cancelledKeys) {
            cancelledKeys.add(k);
        }
    }
 
    public final void close() throws IOException {
        boolean open = selectorOpen.getAndSet(false);
        if (!open)
            return;
        implCloseSelector();// 只有在Selector未关闭的情况下调用,并且只能被调用一次。
    }
    // 关闭Selector
    // 这个方法被close方法调用去执行Selector的关闭操作,只有在Selector未关闭的情况下调用,并且只能被调用一次。具体参考上面close实现
    protected abstract void implCloseSelector() throws IOException;
 
    public final boolean isOpen() {
        return selectorOpen.get();
    }
 
    public final SelectorProvider provider() {
        return provider;
    }
 
    protected final Set<SelectionKey> cancelledKeys() {
        return cancelledKeys;
    }
 
    // 为Selector注册Channel,这个方法被AbstractSelectableChannel.register方法调用
    protected abstract SelectionKey register(AbstractSelectableChannel ch,
                                             int ops, Object att);
 
    protected final void deregister(AbstractSelectionKey key) {
        ((AbstractSelectableChannel)key.channel()).removeKey(key);
    }
 
 
    // -- Interruption machinery --
 
    private Interruptible interruptor = null;
    // 支持异步关闭和中断的begin和end方法
    protected final void begin() {
        if (interruptor == null) {
            interruptor = new Interruptible() {
                    public void interrupt(Thread ignore) {
                        AbstractSelector.this.wakeup();
                    }};
        }
        AbstractInterruptibleChannel.blockedOn(interruptor);
        Thread me = Thread.currentThread();
        if (me.isInterrupted())
            interruptor.interrupt(me);
    }
 
    protected final void end() {
        AbstractInterruptibleChannel.blockedOn(null);
    }
 
}  

SelectorImpl

SelectorImplSelector 的一个实现类,它通常不会被应用程序直接使用,而是通过 Selector.open() 方法获取一个 Selector 实例,这个实例内部可能是一个 SelectorImpl

以下是 SelectorImpl 的一些关键功能:

  1. 注册通道:通过 register() 方法,可以将一个通道(Channel)注册到选择器(Selector)上,并指定感兴趣的操作集(如 SelectionKey.OP_READSelectionKey.OP_WRITE 等)。
  2. 选择操作select() 方法允许选择器等待,直到至少有一个已注册的通道准备好进行感兴趣的操作。当 select() 方法返回时,可以通过 selectedKeys() 方法获取一个包含已就绪通道的 SelectionKey 集合。
  3. 处理已就绪的通道:一旦通过 select() 方法得知哪些通道已就绪,就可以遍历 selectedKeys() 返回的集合,并对每个已就绪的通道进行相应的处理。
  4. 取消注册和关闭:可以通过 SelectionKeycancel() 方法取消通道的注册,也可以通过 close() 方法关闭选择器。

需要注意的是,SelectorImpl 的具体实现因 Java 的不同版本和不同的操作系统而有所不同。因此,在编写依赖于 SelectorImpl 的代码时,应该尽量使用 SelectorSelectionKey 等抽象接口,以确保代码的兼容性和可移植性。

public abstract class SelectorImpl extends AbstractSelector {
    protected Set<SelectionKey> selectedKeys = new HashSet();
    protected HashSet<SelectionKey> keys = new HashSet();
    private Set<SelectionKey> publicKeys;
    private Set<SelectionKey> publicSelectedKeys;

    protected SelectorImpl(SelectorProvider var1) {
        super(var1);
        if (Util.atBugLevel("1.4")) {
            this.publicKeys = this.keys;
            this.publicSelectedKeys = this.selectedKeys;
        } else {
            this.publicKeys = Collections.unmodifiableSet(this.keys);
            this.publicSelectedKeys = Util.ungrowableSet(this.selectedKeys);
        }

    }

    public Set<SelectionKey> keys() {
        if (!this.isOpen() && !Util.atBugLevel("1.4")) {
            throw new ClosedSelectorException();
        } else {
            return this.publicKeys;
        }
    }

    public Set<SelectionKey> selectedKeys() {
        if (!this.isOpen() && !Util.atBugLevel("1.4")) {
            throw new ClosedSelectorException();
        } else {
            return this.publicSelectedKeys;
        }
    }

    protected abstract int doSelect(long var1) throws IOException;

    private int lockAndDoSelect(long var1) throws IOException {
        synchronized(this) {
            if (!this.isOpen()) {
                throw new ClosedSelectorException();
            } else {
                int var10000;
                synchronized(this.publicKeys) {
                    synchronized(this.publicSelectedKeys) {
                        var10000 = this.doSelect(var1);
                    }
                }

                return var10000;
            }
        }
    }

    public int select(long var1) throws IOException {
        if (var1 < 0L) {
            throw new IllegalArgumentException("Negative timeout");
        } else {
            return this.lockAndDoSelect(var1 == 0L ? -1L : var1);
        }
    }

    public int select() throws IOException {
        return this.select(0L);
    }

    public int selectNow() throws IOException {
        return this.lockAndDoSelect(0L);
    }

    public void implCloseSelector() throws IOException {
        this.wakeup();
        synchronized(this) {
            synchronized(this.publicKeys) {
                synchronized(this.publicSelectedKeys) {
                    this.implClose();
                }
            }

        }
    }

    protected abstract void implClose() throws IOException;

    public void putEventOps(SelectionKeyImpl var1, int var2) {
    }

    protected final SelectionKey register(AbstractSelectableChannel var1, int var2, Object var3) {
        if (!(var1 instanceof SelChImpl)) {
            throw new IllegalSelectorException();
        } else {
            SelectionKeyImpl var4 = new SelectionKeyImpl((SelChImpl)var1, this);
            var4.attach(var3);
            synchronized(this.publicKeys) {
                this.implRegister(var4);
            }

            var4.interestOps(var2);
            return var4;
        }
    }

    protected abstract void implRegister(SelectionKeyImpl var1);

    void processDeregisterQueue() throws IOException {
        Set var1 = this.cancelledKeys();
        synchronized(var1) {
            if (!var1.isEmpty()) {
                Iterator var3 = var1.iterator();

                while(var3.hasNext()) {
                    SelectionKeyImpl var4 = (SelectionKeyImpl)var3.next();

                    try {
                        this.implDereg(var4);
                    } catch (SocketException var11) {
                        throw new IOException("Error deregistering key", var11);
                    } finally {
                        var3.remove();
                    }
                }
            }

        }
    }

    protected abstract void implDereg(SelectionKeyImpl var1) throws IOException;

    public abstract Selector wakeup();
}

WindowsSelectorImpl

//poll数组和channel数组的初始容量
private final int INIT_CAP = 8;
//select操作时,每个线程处理的最大FD数量。为INIT_CAP乘以2的幂
private final static int MAX_SELECTABLE_FDS = 1024;
//由这个选择器服务的SelectableChannel的列表
private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[INIT_CAP];
//存放所有FD的包装器,主要用于poll操作
private PollArrayWrapper pollWrapper;
//注册到当前选择器上总的通道数量,初始化为1是因为实例化选择器时加入了wakeupSourceFd
private int totalChannels = 1;
//选择操作所需要的辅助线程数量。每增加一组MAX_SELECTABLE_FDS - 1个通道,就需要一个线程。
private int threadsCount = 0;
//辅助线程列表
private final List<SelectThread> threads = new ArrayList();
//创建一个Pipe实例,用于实现唤醒选择器的功能
private final Pipe wakeupPipe ;
//管道的read端FD,用于实现唤醒选择器的功能
private final int wakeupSourceFd;
//管道的write端FD,用于实现唤醒选择器的功能
private final int wakeupSinkFd;
//关闭锁,通常在注册、注销,关闭,修改选择键的interestOps时都存在竞态条件,主要保护channelArray、pollWrapper等
private Object closeLock = new Object();
//FD为键,SelectionKeyImpl为value的内部map,方便通过FD查找SelectionKeyImpl
private final FdMap fdMap = new FdMap();
//内部类SubSelector中封装了发起poll调用和处理poll调用结果的细节。由主线程调用
private final SubSelector subSelector = new SubSelector();
//选择器每次选择的超时参数
private long timeout;
//中断锁,用于保护唤醒选择器使用的相关竞态资源,如interruptTriggered
private final Object interruptLock = new Object();
//是否触发中断,唤醒选择器的重要标志,由interruptLock保护
private volatile boolean interruptTriggered = false;
//启动锁,当使用多线程处理选择器上Channel的就绪事件时,用于协调这些线程向内核发起系统调用
//辅助线程会在该锁上等待
private final WindowsSelectorImpl.StartLock startLock = new WindowsSelectorImpl.StartLock();
//完成锁,当使用多线程处理选择器上Channel的就绪事件时,用于协调这些线程从系统调用中返回
//主线程会在该锁上等待
private final WindowsSelectorImpl.FinishLock finishLock = new WindowsSelectorImpl.FinishLock();
//updateSelectedKeys调用计数器
//SubSelector.fdsMap中的每个条目都有一个的updateCount值。调用processFDSet时,当我们增加numKeysUpdated,
//会同步将updateCount设置为当前值。 这用于避免多次计算同一个选择键更新多次numKeysUpdated。
//同一个选择键可能出现在readfds和writefds中。
private long updateCount = 0L;

三种SelectionKey集合

  • 在Java NIO中,Selector对象维护了三种与SelectionKey相关的集合,这些集合在Selector的生命周期中扮演着重要角色。这三种集合分别是:
  1. 键集(Key Set)

    • 这个集合包含了所有注册到当前Selector对象的通道的SelectionKey对象。换句话说,每当一个通道通过register()方法注册到Selector时,都会生成一个与之对应的SelectionKey,并添加到这个键集中。这个集合可以通过调用Selectorkeys()方法获得。
    • 需要注意的是,键集是包含了所有注册到选择器的通道的集合,无论这些通道的事件是否就绪。
  2. 已选择键集(Selected Key Set)

    • 当调用Selectorselect()方法时,它会阻塞等待,直到至少有一个注册到它的通道上的某个事件变得就绪(例如,可读、可写等)。一旦有事件就绪,这些事件对应的SelectionKey就会被自动加入到已选择键集中。
    • 这个集合包含了上一次Selector选择期间,发生了就绪事件的通道的SelectionKey对象集合。它是键集的子集,可以通过调用SelectorselectedKeys()方法获得。
    • 在处理完一个SelectionKey后,通常需要从已选择键集中移除它,以避免重复处理。
  3. 已取消键集(Cancelled Key Set)

    • 这个集合包含了那些已经被调用cancel()方法取消的SelectionKey对象,但是关联的通道还没有被撤销。这个集合不能直接获得,并且它始终是键集的子集。
    • 当一个SelectionKey被取消后,它并不会立即从键集中移除,而是会被加入到已取消键集中。在下次调用select()方法时,这些已取消的键会被从键集中移除。

这三种集合共同协作,使得Selector能够高效地管理多个通道的事件,并通过非阻塞的方式处理这些事件。开发者可以通过操作这些集合来监控通道的状态,处理就绪的事件,以及管理通道的注册和取消操作。文章来源地址https://www.toymoban.com/news/detail-846052.html

到了这里,关于Java NIO Selector选择器源码分析的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Java网络编程-深入理解BIO、NIO

    BIO BIO 为 Blocked-IO(阻塞 IO),在 JDK1.4 之前建立网络连接时,只能使用 BIO 使用 BIO 时,服务端会对客户端的每个请求都建立一个线程进行处理,客户端向服务端发送请求后,先咨询服务端是否有线程响应,如果没有就会等待或者被拒绝 BIO 基本使用代码: 服务端: 客户端:

    2024年02月04日
    浏览(29)
  • BIO、NIO、IO多路复用模型详细介绍&Java NIO 网络编程

    上文介绍了网络编程的基础知识,并基于 Java 编写了 BIO 的网络编程。我们知道 BIO 模型是存在巨大问题的,比如 C10K 问题,其本质就是因其阻塞原因,导致如果想要承受更多的请求就必须有足够多的线程,但是足够多的线程会带来内存占用问题、CPU上下文切换带来的性能问题

    2024年02月14日
    浏览(34)
  • Java网络编程----通过实现简易聊天工具来聊聊NIO

    前文我们说过了BIO,今天我们聊聊NIO。 NIO 是什么?NIO官方解释它为 New lO ,由于其特性我们也称之为,Non-Blocking IO。这是jdk1.4之后新增的一套IO标准。 为什么要用NIO呢? 我们再简单回顾下BIO: 阻塞式IO,原理很简单,其实就是多个端点与服务端进行通信时,每个客户端有一个

    2024年02月05日
    浏览(48)
  • Java-NIO篇章(4)——Selector选择器详解

    选择器(Selector)是什么呢?选择器和通道的关系又是什么?这里详细说明,假设不用选择器,那么一个客户端请求数据传输那就需要建立一个连接,为了避免线程阻塞,那么每个客户端开辟一个线程。而学过JVM的都知道,默认每开一个线程需要栈空间内存1MB大小。如果这时

    2024年01月21日
    浏览(27)
  • NIO基础 - 网络编程

    non-blocking io 非阻塞 IO 1.1 Channel Buffer channel 有一点类似于 stream,它就是读写数据的 双向通道 ,可以从 channel 将数据读入 buffer,也可以将 buffer 的数据写入 channel,而之前的 stream 要么是输入,要么是输出,channel 比 stream 更为底层 常见的 Channel 有 FileChannel DatagramChannel SocketCh

    2024年02月02日
    浏览(30)
  • 【网络编程】Volley使用与源码分析

    在2013年Google I/O大会上推出了一个新的网络通信框架Volley。Volley既可以访问网络取得数据,也可以加载图片,并且在性能方面也进行了大幅度的调整,它的设计目标就是非常适合去进行数据量不大,但通信频繁的网络操作,而对于大数据量的网络操作,比如说下载文件等,V

    2024年02月20日
    浏览(22)
  • 10.NIO 网络编程应用实例-群聊系统

    需求:进一步理解 NIO 非阻塞网络编程机制,实现多人群聊 编写一个 NIO 群聊系统,实现客户端与客户端的通信需求(非阻塞) 服务器端:可以监测用户上线,离线,并实现消息转发功能 客户端:通过 channel 可以无阻塞发送消息给其它所有客户端用户,同时可以接受其它客户端用

    2024年02月15日
    浏览(32)
  • 由浅入深Netty基础知识NIO网络编程

    阻塞模式下,相关方法都会导致线程暂停 ServerSocketChannel.accept 会在没有连接建立时让线程暂停 SocketChannel.read 会在没有数据可读时让线程暂停 阻塞的表现其实就是线程暂停了,暂停期间不会占用 cpu,但线程相当于闲置 单线程下,阻塞方法之间相互影响,几乎不能正常工作,

    2024年02月05日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包