文章目录

前言Selector类结构Selector抽象类AbstractSelectorSelectorImplWindowsSelectorImpl三种SelectionKey集合

前言

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

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 keys();

// 返回selected-key集合,key可以直接移除,但是不可以直接增加。并且不是线程安全的集合。

public abstract Set 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() {

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 cancelledKeys = new HashSet();

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 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

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

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

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

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

public abstract class SelectorImpl extends AbstractSelector {

protected Set selectedKeys = new HashSet();

protected HashSet keys = new HashSet();

private Set publicKeys;

private Set 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 keys() {

if (!this.isOpen() && !Util.atBugLevel("1.4")) {

throw new ClosedSelectorException();

} else {

return this.publicKeys;

}

}

public Set 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 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的生命周期中扮演着重要角色。这三种集合分别是:

键集(Key Set):

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

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

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

这三种集合共同协作,使得Selector能够高效地管理多个通道的事件,并通过非阻塞的方式处理这些事件。开发者可以通过操作这些集合来监控通道的状态,处理就绪的事件,以及管理通道的注册和取消操作。

好文链接

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: