文章目录
- 前言
- Selector类结构
- Selector抽象类
- AbstractSelector
- SelectorImpl
- WindowsSelectorImpl
- 三种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<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
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<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的生命周期中扮演着重要角色。这三种集合分别是：
-
键集（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能够高效地管理多个通道的事件，并通过非阻塞的方式处理这些事件。开发者可以通过操作这些集合来监控通道的状态，处理就绪的事件，以及管理通道的注册和取消操作。
