Spring架构篇--2.5 远程通信基础Select 源码篇--window--Select.open()

前言：在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())；