前言:在Socket通信中使用Select 来对NIO 进行实现,那么它们的实现方式是怎样的呢,本文从 Selector.open() 进行第一步的分析;
Selector.open() :
Selector 类:
public static Selector open() throws IOException {// 通过 SelectorProvider.provider() 获取SelectorProvider实例// 通过openSelector() 获取不同系统的Selector 实现return SelectorProvider.provider().openSelector();}
先看:SelectorProvider.provider():
SelectorProvider 类:
private static final Object lock = new Object();
private static SelectorProvider provider = null;public static SelectorProvider provider() {synchronized (lock) {// 通过 lock 对象锁,保证当前进程只会有一个SelectorProvider 对象if (provider != null)// 如果发现进程中已经实例化过SelectorProvider 对象则直接返回return provider;// 进程下没有过SelectorProvider 对象则进行加载return AccessController.doPrivileged(new PrivilegedAction<SelectorProvider>() {public SelectorProvider run() {if (loadProviderFromProperty())return provider;if (loadProviderAsService())return provider;provider = sun.nio.ch.DefaultSelectorProvider.create();return provider;}});}
}
SelectorProvider.provider()方法会在System Property中不存在java.nio.channels.spi.SelectorProvider属性和不能找到SelectorProvider的实现类时,创建一个默认的sun.nio.ch.DefaultSelectorProvider来作为SelectorProvide;DefaultSelectorProvider 会根据不同的操作系统返回:
window 下的DefaultSelectorProvider:
package sun.nio.ch;import java.nio.channels.spi.SelectorProvider;public class DefaultSelectorProvider {private DefaultSelectorProvider() {}public static SelectorProvider create() {return new WindowsSelectorProvider();}
}
DefaultSelectorProvider.create() 创建方法:可以看到返回了WindowsSelectorProvider一个实例对象;
package sun.nio.ch;import java.io.IOException;
import java.nio.channels.spi.AbstractSelector;public class WindowsSelectorProvider extends SelectorProviderImpl {public WindowsSelectorProvider() {}public AbstractSelector openSelector() throws IOException {return new WindowsSelectorImpl(this);}
}
在回到Selector 中的 SelectorProvider.provider().openSelector(), SelectorProvider.provider() 实际上返回了WindowsSelectorProvider一个实例对象,然后调用WindowsSelectorProvider的openSelector()方法,可以看到这里创建了一个WindowsSelectorImpl 实例对象并进行了返回:
这里看下对象的关系:
1 ) WindowsSelectorImpl extends SelectorImpl;
2) SelectorImpl extends AbstractSelector;
3) AbstractSelector extends Selector
4) Selector implements Closeable
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;WindowsSelectorImpl(SelectorProvider var1) throws IOException {// 调用 SelectorImpl 的父类构造方法 // 将 上一步WindowsSelectorProvider的openSelector() SelectorProvider 实例传入super(var1);// Pipe wakeupPipe = Pipe.open() 的fd 文件描述符赋值this.wakeupSourceFd = ((SelChImpl)this.wakeupPipe.source()).getFDVal();SinkChannelImpl var2 = (SinkChannelImpl)this.wakeupPipe.sink();// 禁用 Nagle 算法,当 sink 端写入 1 字节数据时,将立即发送,而不必等到将较小的包组合成较大的包再发送,// 这样 source 端就可以立马读取数据var2.sc.socket().setTcpNoDelay(true);// Pipe wakeupPipe = Pipe.open() 的fd 文件描述符赋值this.wakeupSinkFd = var2.getFDVal();// Pipe wakeupPipe = Pipe.open() 的fd 文件描述符赋值加入到管道数组中// 把唤醒端的文件描述符(wakeupSourceFd)放到pollWrapper里this.pollWrapper.addWakeupSocket(this.wakeupSourceFd, 0);
}
WindowsSelectorImpl 类中通过socket 建立了管道连接,并将管道符进行保存;
super(var1) 调用 SelectorImpl 的父类构造方法:
protected Set<SelectionKey> selectedKeys = new HashSet();
protected HashSet<SelectionKey> keys = new HashSet();
private Set<SelectionKey> publicKeys;
private Set<SelectionKey> publicSelectedKeys;
protected SelectorImpl(SelectorProvider var1) {// 调用 AbstractSelector 的父类构造方法super(var1);// publicKeys 和 publicSelectedKeys 进行映射赋值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);}}
SelectorImpl 的构造,对publicKeys和publicSelectedKeys 集合做了映射,方便操作selectedKeys和keys 集合可以直接影响数据;
super(var1) 调用 AbstractSelector的父类构造方法:
private final SelectorProvider provider;
protected AbstractSelector(SelectorProvider provider) {this.provider = provider;
}
因为WindowsSelectorImpl 的实例处理构造方法的调用就是对Pipe wakeupPipe = Pipe.open() 产生的文件操作符进行赋值,这里看下 Pipe.open():
Pipe 类:
public static Pipe open() throws IOException {// SelectorProvider.provider() 返回直接创建好的WindowsSelectorProvider的openSelector() SelectorProvider 实例传入// 然后调用openPipe()return SelectorProvider.provider().openPipe();
}
SelectorProviderImpl类中的openPipe():
public Pipe openPipe() throws IOException {// 返回 PipeImpl 的实例,传入创建好的WindowsSelectorProvider的openSelector() SelectorProvider 实例return new PipeImpl(this);
}
PipeImpl 的构造方法:
private static final int NUM_SECRET_BYTES = 16;
private static final Random RANDOM_NUMBER_GENERATOR = new SecureRandom();
private SourceChannel source;
private SinkChannel sink;PipeImpl(SelectorProvider var1) throws IOException {try {// AccessController.doPrivileged() 借用权限完成方法的执行AccessController.doPrivileged(new PipeImpl.Initializer(var1));} catch (PrivilegedActionException var3) {throw (IOException)var3.getCause();}
}
重点看下PipeImpl 内部类中 new PipeImpl.Initializer(var1),在进行构造之后,执行run方法:
private class Initializer implements PrivilegedExceptionAction<Void> {private final SelectorProvider sp;private IOException ioe;private Initializer(SelectorProvider var2) {// 异常原因this.ioe = null;// SelectorProvider 实例this.sp = var2;}public Void run() throws IOException {// 启动线程执行LoopbackConnector 的run方法PipeImpl.Initializer.LoopbackConnector var1 = new PipeImpl.Initializer.LoopbackConnector();var1.run();if (this.ioe instanceof ClosedByInterruptException) {// 如果建立通道过程中发生了关闭异常则重新发起线程调用LoopbackConnector 的run方法this.ioe = null;Thread var2 = new Thread(var1) {public void interrupt() {}};var2.start();while(true) {try {var2.join();break;} catch (InterruptedException var4) {}}Thread.currentThread().interrupt();}if (this.ioe != null) {// 发生异常抛出异常throw new IOException("Unable to establish loopback connection", this.ioe);} else {return null;}}private class LoopbackConnector implements Runnable {private LoopbackConnector() {}public void run() {// 声明服务端的 ServerSocketChannel ServerSocketChannel var1 = null;// 声明客户端的SocketChannel SocketChannel var2 = null;SocketChannel var3 = null;try {// 声明var4 和 var5 两个内存空间分别为16个字节大小ByteBuffer var4 = ByteBuffer.allocate(16);ByteBuffer var5 = ByteBuffer.allocate(16);// 声明本机的地址InetAddress var6 = InetAddress.getByName("127.0.0.1");assert var6.isLoopbackAddress();// 声明var7 地址InetSocketAddress var7 = null;while(true) {if (var1 == null || !var1.isOpen()) {// 初始化服务端的ServerSocketChannel var1 = ServerSocketChannel.open();// 绑定服务端监听的端口ip 和端口var1.socket().bind(new InetSocketAddress(var6, 0));// 将地址赋值给var7 var7 = new InetSocketAddress(var6, var1.socket().getLocalPort());}// 打开客户端SocketChannel ,地址为var7 建立连接var2 = SocketChannel.open(var7);// 向var4 写入随机数PipeImpl.RANDOM_NUMBER_GENERATOR.nextBytes(var4.array());do {// 向SocketChannel 建立的连接写入数据var2.write(var4);} while(var4.hasRemaining());// 返回此缓冲区var4.rewind();// 服务端阻塞等待连接连接var3 = var1.accept();do {// 从服务端连接中读取数据var3.read(var5);} while(var5.hasRemaining());// 返回此缓冲区var5.rewind();// 如果客户端发送的数据和服务端接收到的数据相同说通道完成建立if (var5.equals(var4)) {// 放入通道描述PipeImpl.this.source = new SourceChannelImpl(Initializer.this.sp, var2);PipeImpl.this.sink = new SinkChannelImpl(Initializer.this.sp, var3);// 跳出循环break;}// 关闭 var3.close();var2.close();}} catch (IOException var18) {try {if (var2 != null) {var2.close();}if (var3 != null) {var3.close();}} catch (IOException var17) {}// 异常原因Initializer.this.ioe = var18;} finally {try {if (var1 != null) {// 关闭var1.close();}} catch (IOException var16) {}}}}
}
- windows下的实现是创建两个本地的socketChannel,然后连接(链接的过程通过写一个随机long做两个socket的链接校验),两个socketChannel分别实现了管道的source与sink端。
- source端由前面提到的WindowsSelectorImpl放到了pollWrapper中(pollWrapper.addWakeupSocket(wakeupSourceFd, 0));
- 改通道的建立并不是为了确保连接,而是为了后续Select在获取内核准备好的数据时,一旦有socket返回可读/可写时间,方便唤醒对主线程的唤醒使用;
pollWrapper.addWakeupSocket(this.wakeupSourceFd, 0):
PollArrayWrapper类:
void putDescriptor(int var1, int var2) {this.pollArray.putInt(SIZE_POLLFD * var1 + 0, var2);
}void putEventOps(int var1, int var2) {this.pollArray.putShort(SIZE_POLLFD * var1 + 4, (short)var2);
}void addWakeupSocket(int var1, int var2) {this.putDescriptor(var2, var1);this.putEventOps(var2, Net.POLLIN);
}
这里将source的POLLIN事件标识为感兴趣的,当sink端有数据写入时,source对应的文件描述符wakeupSourceFd就会处于就绪状态;
总结:
- window–Select.open() 通过加锁的方式获取进程下的唯一一个 WindowsSelectorImpl对象实例。
- WindowsSelectorImpl 的对象实例中通过 对127.0.0.1 端口为0 ,建立两个连接的方式拿到文件操作符,并把 把唤醒端的文件描述符(wakeupSourceFd)放到pollWrapper里;
参考:
Java NIO 之 Selector(第一部分Selector.open());