NIO（四）：源码分析_Register部分

我就是我 2023-06-27 05:42 14阅读 0赞

*  注册，就是将当前`Channel`注册到`Selector`上，是NIO源码的核心部分

# 1，Channel.register()部分 #

*  `serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT)`：注册事件

public final SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException {
        // 用注册锁保证同步
        synchronized (regLock) {
            if (!isOpen())
                throw new ClosedChannelException();
            if ((ops & ~validOps()) != 0)
                throw new IllegalArgumentException();
            if (blocking)
                throw new IllegalBlockingModeException();
            // 从当前Channel自身的注册列表中获取数据
            SelectionKey k = findKey(sel);
            // 获取到数据后，只对SelectionKey状态进行变更
            if (k != null) {
                k.interestOps(ops);
                k.attach(att);
            }
            // key为空，说明该任务为新任务，加锁后注册
            if (k == null) {
                synchronized (keyLock) {
                    if (!isOpen())
                        throw new ClosedChannelException();
                    k = ((AbstractSelector)sel).register(this, ops, att);
                    // 注册完成后，将新注册的SelectionKey添加的Channel的注册列表中
                    addKey(k);
                }
            }
            return k;
        }
    }

*  `AbstractSelectableChannel.findKey()`：从`Channel`的`SelectionKey[]`数组中获取存在数据

private SelectionKey findKey(Selector sel) {
        synchronized (keyLock) {
            if (keys == null)
                return null;
            for (int i = 0; i < keys.length; i++)
                // 一个Channel跟一个Selector是强绑定的,可以通过Selector匹配获取注册的Key
                // Channel和Selector之间是多对多的关系
                if ((keys[i] != null) && (keys[i].selector() == sel))
                    return keys[i];
            return null;
        }
    }

*  如果 Channel 的注册列表中不存在该数据，则直接转发到 Selector 进行该 Channel 注册
 *  `AbstractSelectableChannel.addKey()` ：注册完成后，添加到 Channel 的注册列表中

private void addKey(SelectionKey k) {
        assert Thread.holdsLock(keyLock);
        int i = 0;
        // 首先数据已经初始化且未满， 正常填充
        if ((keys != null) && (keyCount < keys.length)) {
            for (i = 0; i < keys.length; i++)
                if (keys[i] == null)
                    break;
        // 数组未初始化，直接初始化，并制定默认长度3
        } else if (keys == null) {
            keys =  new SelectionKey[3];
        // 数组已经初始化，且数据已满，
        // 进行二倍扩容和原始数据迁移
        } else {
            // Grow key array
            int n = keys.length * 2;
            SelectionKey[] ks =  new SelectionKey[n];
            for (i = 0; i < keys.length; i++)
                ks[i] = keys[i];
            keys = ks;
            i = keyCount;
        }
        // 获取到可以填充的下标，进行数据填充
        keys[i] = k;
        keyCount++;
    }

# 2，Selector.register()部分 #

*  `((AbstractSelector)sel).register(this, ops, att)`：通道未注册到选择器，进行注册

protected final SelectionKey register(AbstractSelectableChannel var1, int var2, Object var3) {
        if(!(var1 instanceof SelChImpl)) {
            throw new IllegalSelectorException();
        } else {
            // 初始化一个注册器，封装通道和选择器
            // 生成的对象真是实例为SelectionKeyImpl
            SelectionKeyImpl var4 = new SelectionKeyImpl((SelChImpl)var1, this);
            // 附加各位属性, 会透传到下一个事件
            var4.attach(var3);
            Set var5 = this.publicKeys;
            // 对注册的SelectionKey列表进行同步处理
            synchronized(this.publicKeys) {
                // 初始化完成后，进行最终注册,添加Socket句柄
                this.implRegister(var4);
            }
    		// 添加事件类型, 即修改原事件状态
            var4.interestOps(var2);
            return var4;
        }
    }
    
    SelectionKeyImpl(SelChImpl var1, SelectorImpl var2) {
        this.channel = var1;
        this.selector = var2;
    }

*  `WindowsSelectorImpl.implRegister`：注册

// 注册
    protected void implRegister(SelectionKeyImpl var1) {
        Object var2 = this.closeLock;
        synchronized(this.closeLock) {
            if(this.pollWrapper == null) {
                throw new ClosedSelectorException();
            } else {
                // 对SelectionKeyImpl[] channelArray数组进行二倍扩容, 默认长度为8
                // 没增加1024个Channel,则增加一个线程处理
                this.growIfNeeded();
                // 填充到数据的下一个索引位置
                this.channelArray[this.totalChannels] = var1;
                // 设置SelectionKey的索引值, 添加事件时候会用到
                var1.setIndex(this.totalChannels);
                // SelectionKey句柄和对象的映射
                this.fdMap.put(var1);
                // 添加到选择器的注册列表中
                this.keys.add(var1);
                // 添加Socket句柄到pollWrapper
                // 注册此处传递的总数,也就代表当前注册对象索引
                this.pollWrapper.addEntry(this.totalChannels, var1);
                // 表示已注册的Channel总数
                ++this.totalChannels;
            }
        }
    }

*  `WindowsSelectorImpl.growIfNeeded`：扩容

private void growIfNeeded() {
        if(this.channelArray.length == this.totalChannels) {
            // 数量超过定长，直接进行二倍扩容
            int var1 = this.totalChannels * 2;
            SelectionKeyImpl[] var2 = new SelectionKeyImpl[var1];
            System.arraycopy(this.channelArray, 1, var2, 1, this.totalChannels - 1);
            this.channelArray = var2;
            this.pollWrapper.grow(var1);
        }
    	// 数量为1024的倍数，对线程加1，select()时作为子线程进行查找事件
        if(this.totalChannels % 1024 == 0) {
            this.pollWrapper.addWakeupSocket(this.wakeupSourceFd, this.totalChannels);
            ++this.totalChannels;
            ++this.threadsCount;
        }
    
    }

*  `PollArrayWrapper.addEntry()`：添加注册对象的Socket句柄到内存对象中

void addEntry(int var1, SelectionKeyImpl var2) {
        this.putDescriptor(var1, var2.channel.getFDVal());
    }
    
    void putDescriptor(int var1, int var2) {
        this.pollArray.putInt(SIZE_POLLFD * var1 + 0, var2);
    }

*  `SelectionKeyImpl.interestOps`：修改注册事件

public SelectionKey nioInterestOps(int var1) {
        if((var1 & ~this.channel().validOps()) != 0) {
            throw new IllegalArgumentException();
        } else {
            // 进行注册事件修改
            // 注册事件修改，主要分析ServerSocketChannel和SocketChannel两部分
            this.channel.translateAndSetInterestOps(var1, this);
            this.interestOps = var1;
            return this;
        }
    }
    
    // ServerSocketChannel
    public void translateAndSetInterestOps(int var1, SelectionKeyImpl var2) {
        int var3 = 0;
        if((var1 & 16) != 0) { // 16 表示accept事件
            var3 |= Net.POLLIN;
        }
    	// 对事件进行替换
        var2.selector.putEventOps(var2, var3);
    }
    
    // SocketChannel
    public void translateAndSetInterestOps(int var1, SelectionKeyImpl var2) {
        int var3 = 0;
        if((var1 & 1) != 0) { // 读事件
            var3 |= Net.POLLIN;
        }
    
        if((var1 & 4) != 0) { // 写事件
            var3 |= Net.POLLOUT;
        }
    
        if((var1 & 8) != 0) { // 连接事件
            var3 |= Net.POLLCONN;
        }
    	// 对事件类型进行替换
        var2.selector.putEventOps(var2, var3);
    }

*  `WindowsSelectorImpl.putEventOps`：事件替换

public void putEventOps(SelectionKeyImpl var1, int var2) {
        Object var3 = this.closeLock;
        synchronized(this.closeLock) {
            if(this.pollWrapper == null) {
                throw new ClosedSelectorException();
            } else {
                // 从注册对象中获取注册事件的索引位置
                int var4 = var1.getIndex();
                if(var4 == -1) {
                    throw new CancelledKeyException();
                } else {
                    // 通过内存对象直接操作地址空间，进行事件类型替换
                    this.pollWrapper.putEventOps(var4, var2);
                }
            }
        }
    }

*  注册源码分析完成，最终注册事件添加到 SelectorImpl.keys 和 AbstractSelectableChannel.keys，并在 AllocatedNativeObject 内存对象中存储，可以通过 select() 进行查找