NIO（五）：源码分析_Select部分

古城微笑少年丶 2023-06-27 05:42 9阅读 0赞

*  之前已经对register源码进行了分析，注册已经是NIO的核心代码，而Select部分更是重中之重
    
     *  注册时注册`Socket`句柄到内存对象中
     *  `select()`时构造多道线程取对应区间的`Socket`句柄，线程在处理过程中分为协调线程和工作线程，注意其中协调获取的关系
     *  添加有效的`Socket`句柄到`SelectionKey`列表中，通过`selectKeys()`可以直接获取到，进行后续事件处理

# 1，Select整体流程部分 #

*  `selector.select()`：调用直接走向`doSelect()`实现
 *  `WindowsSelectorImpl.doSelect()`：通过选择器获取有效事件

protected int doSelect(long var1) throws IOException {
        if(this.channelArray == null) {
            throw new ClosedSelectorException();
        } else {
            this.timeout = var1;
            // 处理掉注销的队列数据
            this.processDeregisterQueue();
            if(this.interruptTriggered) {
                this.resetWakeupSocket();
                return 0;
            } else {
                // 调整线程数量，少增多减
                // 并将新增线程设置为预启动状态
                this.adjustThreadsCount();
                // 重置完成线程数， 数量为当前线程数量
                this.finishLock.reset();
                // 放开所有就绪线程，所有线程检测到一个就绪Socket句柄后返回
                this.startLock.startThreads();
    
                try {
                    this.begin();
                    try {
                        // 每一个线程监听（0~1023）* N区间的Socket句柄
                        // 如果当前没有句柄，会在此处阻塞
                        this.subSelector.poll();
                    } catch (IOException var7) {
                        this.finishLock.setException(var7);
                    }
    				// waitForHelperThreads()用于阻塞直到所有线程执行完毕
                    // 在adjustThreadsCount中，每个线程执行完毕都会等待，所有线程执行完毕后会依次notify()
                    if(this.threads.size() > 0) {
                        this.finishLock.waitForHelperThreads();
                    }
                } finally {
                    this.end();
                }
                this.finishLock.checkForException();
                // 再次检查失效的注册事件
                this.processDeregisterQueue();
                // 更新selectKeys，并返回数量
                int var3 = this.updateSelectedKeys();
                this.resetWakeupSocket();
                return var3;
            }
        }
    }

# 2，Select注销队列事件处理过程 #

*  `processDeregisterQueue()`：失效事件取消代码

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();
                    }
                }
            }
        }
    }
    
    // implDereg()
    protected void implDereg(SelectionKeyImpl var1) throws IOException {
        // 获取索引，也就是在内存空间中的位置
        int var2 = var1.getIndex();
        assert var2 >= 0;
        Object var3 = this.closeLock;
        synchronized(this.closeLock) {
            // 失效的节点不是最后一个，就用最后一个几点进行补充，也就是把数组后面的元素补充要前面
            // 不断充实数组，减少数组扩容次数
            if(var2 != this.totalChannels - 1) {
                SelectionKeyImpl var4 = this.channelArray[this.totalChannels - 1];
                this.channelArray[var2] = var4;
                var4.setIndex(var2);
                // 对内存空间中进行位置替换
                this.pollWrapper.replaceEntry(this.pollWrapper, this.totalChannels - 1, this.pollWrapper, var2);
            }
            // 重置当前索引为无意义索引
            var1.setIndex(-1);
        }
    	// 把数组的最后一位置空，并且对总数和总线程数进行相关匹配处理
        this.channelArray[this.totalChannels - 1] = null;
        --this.totalChannels;
        if(this.totalChannels != 1 && this.totalChannels % 1024 == 1) {
            --this.totalChannels;
            --this.threadsCount;
        }
    
        // 分别从注册事件，句柄Map映射中移除数据
        this.fdMap.remove(var1);
        this.keys.remove(var1);
        this.selectedKeys.remove(var1);
        // 清理Channel中的注册标识
        this.deregister(var1);
        SelectableChannel var7 = var1.channel();
        if(!var7.isOpen() && !var7.isRegistered()) {
            ((SelChImpl)var7).kill();
        }
    }
    
    // AbstractSelectableChannel.removeKey
    void removeKey(SelectionKey k) {                    // package-private
        synchronized (keyLock) {
            // 遍历Channel中的注册数据，将该元素位置置空，长度减一
            for (int i = 0; i < keys.length; i++)
                if (keys[i] == k) {
                    keys[i] = null;
                    keyCount--;
                }
            // 重置失效状态为失效
            ((AbstractSelectionKey)k).invalidate();
        }
    }

# 3，Select事件获取线程准备部分 #

*  `adjustThreadsCount()`：准备事件获取线程

private void adjustThreadsCount() {
        int var1;
        // 线程数初始化为0，也就是如果不足1024注册的话，该部分不会执行
        // 判断线程数，该线程数对应Channel初始化时为1024倍数时的递增
        // 当线程数量不足时，初始化线程，并启动，（此处启动会阻塞）
        if(this.threadsCount > this.threads.size()) {
            for(var1 = this.threads.size(); var1 < this.threadsCount; ++var1) {
                // 初始化线程，此处会初始化多个SelectThread对象
                // 每一个SelectThread内部包含一个SubSelector
                // 一个SubSelector负责的句柄区间就是(0 ~ 1024) * var1
                WindowsSelectorImpl.SelectThread var2 = new WindowsSelectorImpl.SelectThread(var1);
                this.threads.add(var2);
                var2.setDaemon(true);
                // 启动
                var2.start();
            }
        // 线程数量大于有效线程数量时候，将线程失效，唤醒后不会执行
        } else if(this.threadsCount < this.threads.size()) {
            for(var1 = this.threads.size() - 1; var1 >= this.threadsCount; --var1) {
                ((WindowsSelectorImpl.SelectThread)this.threads.remove(var1)).makeZombie();
            }
        }
    }

# 4，Select核心过程相关的 `WindowsSelectorImpl` 四个核心内部类 #

## 4.1，`StartLock` ##

private final class StartLock {
        // 执行select()方法的次数
        private long runsCounter;
        // 每一次启动，对runsCounter递增
        // 并且启动所有阻塞的线程
        private synchronized void startThreads() {
            ++this.runsCounter;
            this.notifyAll();
        }
    
        // 在同一批次执行中,线程等待启动,第一次执行,值都为空
        private synchronized boolean waitForStart(WindowsSelectorImpl.SelectThread var1) {
            while(this.runsCounter == var1.lastRun) {
                try {
                    // 同一批次执行，线程统一为预执行状态，在此阻塞，等待唤醒
                    WindowsSelectorImpl.this.startLock.wait();
                } catch (InterruptedException var3) {
                    Thread.currentThread().interrupt();
                }
            }
            // 被唤醒后,再次根据线程是否失效状态判断是否继续执行下去
            if(var1.isZombie()) {
                return true;
            } else {
                // 执行完成后,因为批次相等,会继续阻塞
                var1.lastRun = this.runsCounter;
                return false;
            }
        }
    }

## 4.2，`FinishLock` ##

private final class FinishLock {
        // 剩余多少线程还没有执行完本次select()统计
        private int threadsToFinish;
        IOException exception;
    
        private FinishLock() {
            this.exception = null;
        }
    
        // 每一次select()完成后,进行标记参数重置
        private void reset() {
            this.threadsToFinish = WindowsSelectorImpl.this.threads.size();
        }
    
        // 在SelectThread中,只有poll到数据,才会走到该方法,不然会一直阻塞
        private synchronized void threadFinished() {
            // 如果要执行的线程数与总的线程数相等, 则唤醒主线程进行工作, 
            // 因为内部有参数控制, 多次唤醒无效,所以此处只唤醒一次
            if(this.threadsToFinish == WindowsSelectorImpl.this.threads.size()) {
                // 唤醒select()主线程开始工作
                // 此处唤醒应该是通过Pipe管道的方式唤醒的
                WindowsSelectorImpl.this.wakeup();
            }
    		// 剩余线程数量递减
            --this.threadsToFinish;
            // 当剩余线程为0时,释放阻塞在该节点的锁
            // 唤醒其他线程继续执行
            if(this.threadsToFinish == 0) {
                this.notify();
            }
        }
    
        // 表示等待sub工作线程执行完成
        private synchronized void waitForHelperThreads() {
            // 唤醒select()主线程
            if(this.threadsToFinish == WindowsSelectorImpl.this.threads.size()) {
                WindowsSelectorImpl.this.wakeup();
            }
            // 如果存在辅助线程没有执行完, 阻塞当前线程, 并等待
            while(this.threadsToFinish != 0) {
                try {
                    WindowsSelectorImpl.this.finishLock.wait();
                } catch (InterruptedException var2) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

## 4.3，`SelectThread` ##

private final class SelectThread extends Thread {
        private final int index;
        // 真正执行select操作的执行器
        final WindowsSelectorImpl.SubSelector subSelector;
        private long lastRun;
        private volatile boolean zombie;
    
        private SelectThread(int var2) {
            this.lastRun = 0L;
            this.index = var2;
            this.subSelector = WindowsSelectorImpl.this.new SubSelector(var2);
            this.lastRun = WindowsSelectorImpl.this.startLock.runsCounter;
        }
    
        void makeZombie() {
            this.zombie = true;
        }
    
        boolean isZombie() {
            return this.zombie;
        }
    
        public void run() {
            // waitForStart(this)：线程阻塞，等待唤醒，唤醒后过滤掉失效线程
            // threadFinished()：唤醒工作线程进行事件选择
            for(; !WindowsSelectorImpl.this.startLock.waitForStart(this); WindowsSelectorImpl.this.finishLock.threadFinished()) {
                try {
                    // sub工作线程阻塞获取注册事件
                    this.subSelector.poll(this.index);
                } catch (IOException var2) {
                    WindowsSelectorImpl.this.finishLock.setException(var2);
                }
            }
        }
    }

## 4.4，SubSelector ##

private final class SubSelector {
        // 当前线程读取的句柄区间
        private final int pollArrayIndex;
        // 事件句柄数组，其中第一个元素表示长度，之后表示真实的句柄地址
        private final int[] readFds;
        private final int[] writeFds;
        private final int[] exceptFds;
    
        private SubSelector() {
            this.readFds = new int[1025];
            this.writeFds = new int[1025];
            this.exceptFds = new int[1025];
            this.pollArrayIndex = 0;
        }
    
        private SubSelector(int var2) {
            this.readFds = new int[1025];
            this.writeFds = new int[1025];
            this.exceptFds = new int[1025];
            this.pollArrayIndex = (var2 + 1) * 1024;
        }
    
        // poll操作后，会将上面三个句柄数组进行填充
        private int poll() throws IOException {
            return this.poll0(WindowsSelectorImpl.this.pollWrapper.pollArrayAddress, Math.min(WindowsSelectorImpl.this.totalChannels, 1024), this.readFds, this.writeFds, this.exceptFds, WindowsSelectorImpl.this.timeout);
        }
    
        private int poll(int var1) throws IOException {
            return this.poll0(WindowsSelectorImpl.this.pollWrapper.pollArrayAddress + (long)(this.pollArrayIndex * PollArrayWrapper.SIZE_POLLFD), Math.min(1024, WindowsSelectorImpl.this.totalChannels - (var1 + 1) * 1024), this.readFds, this.writeFds, this.exceptFds, WindowsSelectorImpl.this.timeout);
        }
    
        private native int poll0(long var1, int var3, int[] var4, int[] var5, int[] var6, long var7);
    
        // processSelectedKeys
        // 此处分别对不同的事件类型进行处理
        private int processSelectedKeys(long var1) {
            byte var3 = 0;
            // 分别对三个句柄数组进行操作，并返回数量
            int var4 = var3 + this.processFDSet(var1, this.readFds, Net.POLLIN, false);
            var4 += this.processFDSet(var1, this.writeFds, Net.POLLCONN | Net.POLLOUT, false);
            var4 += this.processFDSet(var1, this.exceptFds, Net.POLLIN | Net.POLLCONN | Net.POLLOUT, true);
            // 最终返回整体数量
            return var4;
        }
    
        // processFDSet
        private int processFDSet(long var1, int[] var3, int var4, boolean var5) {
            int var6 = 0;
    		// 从此处可以看出，第一个元素表示数量
            // 从第二个元素开始通过句柄地址获取数据进行处理
            for(int var7 = 1; var7 <= var3[0]; ++var7) {
                int var8 = var3[var7];
                if(var8 == WindowsSelectorImpl.this.wakeupSourceFd) {
                    synchronized(WindowsSelectorImpl.this.interruptLock) {
                        WindowsSelectorImpl.this.interruptTriggered = true;
                    }
                } else {
                    // 通过句柄地址获取到的事件数据
                    WindowsSelectorImpl.MapEntry var9 = WindowsSelectorImpl.this.fdMap.get(var8);
                    if(var9 != null) {
                        SelectionKeyImpl var10 = var9.ski;
                        if(!var5 || !(var10.channel() instanceof SocketChannelImpl) || !WindowsSelectorImpl.this.discardUrgentData(var8)) {
                            // 此处表示注册的事件在列表中已经存在,对事件类型进行变更
                            if(WindowsSelectorImpl.this.selectedKeys.contains(var10)) {
                                if(var9.clearedCount != var1) {
                                    if(var10.channel.translateAndSetReadyOps(var4, var10) && var9.updateCount != var1) {
                                        var9.updateCount = var1;
                                        ++var6;
                                    }
                                } else if(var10.channel.translateAndUpdateReadyOps(var4, var10) && var9.updateCount != var1) {
                                    var9.updateCount = var1;
                                    ++var6;
                                }
    
                                var9.clearedCount = var1;
                            } else {
                                // 如果不存在,则添加到集合中去
                                if(var9.clearedCount != var1) {
                                    var10.channel.translateAndSetReadyOps(var4, var10);
                                    if((var10.nioReadyOps() & var10.nioInterestOps()) != 0) {
                                        WindowsSelectorImpl.this.selectedKeys.add(var10);
                                        var9.updateCount = var1;
                                        ++var6;
                                    }
                                } else {
                                    var10.channel.translateAndUpdateReadyOps(var4, var10);
                                    if((var10.nioReadyOps() & var10.nioInterestOps()) != 0) {
                                        WindowsSelectorImpl.this.selectedKeys.add(var10);
                                        var9.updateCount = var1;
                                        ++var6;
                                    }
                                }
    
                                var9.clearedCount = var1;
                            }
                        }
                    }
                }
            }
    
            return var6;
        }
    }

## 4.5，四个核心类调用关系 ##

*  在 adjustThreadsCount() 中进行线程准备时，虽然对初始化的线程类 SelectThread 进行了 start() 操作，但是在线程内的 run() 中进行了线程阻塞 startLock.waitForStart()，等待再次唤醒；此外，线程数量是在注册时 SelectionKey 超过 1024的倍数时进行递增的
 *  线程准备后，将已经完成线程的计数器进行重置（finishLock.reset()），重置为线程数量，表示所有线程已经启动，并都没有执行完成
 *  重置完成后，唤醒所有准备好的线程 notifyAll() 执行 poll() 操作，获取有效事件
 *  此处通过 poll() 获取数据，各个线程以 (0 ~ 1023) \* N 为区间方式，分别取获取到Socket句柄位置后添加的 SubSelector 的成员数组中
 *  poll() 数据的工作线程线程执行完成后，会调用一次 wakeUp() 方法唤醒 select() 主线程进行工作，因为 wakeUp() 多次调用无效，所以只有第一个完成的工作线程才会进行唤醒操作
 *  poll() 数据的工作线程完成后，下一步会继续判断所有工作线程是否已经全部完成（this.threads.size() > 0），如果没有全部完成，则线程进行等待（this.finishLock.waitForHelperThreads()），等最后一道工作线程（this.threadsToFinish == 0）执行完成后，会唤醒该线程继续执行（this.notify()）
 *  所有工作线程执行完成后，则当前所有已经注册事件的Socket句柄已经被各个 SubSelector 持有，下一步则从根据各个 SubSelector 持有的 Socket 句柄索引，获取事件，并添加到 selectedKeys 中，等待后续业务处理

# 5，添加数据到集合中 #

*  `updateSelectedKeys()`：添加选择到的数据到集合中

private int updateSelectedKeys() {
        ++this.updateCount;
        byte var1 = 0;
        int var4 = var1 + this.subSelector.processSelectedKeys(this.updateCount);
    
        WindowsSelectorImpl.SelectThread var3;
        // 遍历每一个线程数据进行处理
        for(Iterator var2 = this.threads.iterator(); var2.hasNext(); var4 += var3.subSelector.processSelectedKeys(this.updateCount)) {
            var3 = (WindowsSelectorImpl.SelectThread)var2.next();
        }
    
        return var4;
    }