java nio 源码分析1 事件注册

public static Selector open() throws IOException {
  return SelectorProvider.provider().openSelector();
}

public AbstractSelector openSelector() throws IOException {
    return new EPollSelectorImpl(this);
}

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();
        SelectionKey k = findKey(sel);  // 根据Selector找到SelectionKey
        if (k != null) {
            k.interestOps(ops);  // 更新注册事件
            k.attach(att);
        }
        if (k == null) {
            // New registration
            synchronized (keyLock) {
                if (!isOpen())
                    throw new ClosedChannelException();
                k = ((AbstractSelector)sel).register(this, ops, att);  // 关键, 注册事件
                addKey(k);  // 保存该SelectionKey
            }
        }
        return k;
    }
}

// SelectorImpl.java
protected final SelectionKey register(AbstractSelectableChannel ch,
                                      int ops,
                                      Object attachment)
{
    if (!(ch instanceof SelChImpl))
        throw new IllegalSelectorException();
    SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);  // 创建SelectionKey
    k.attach(attachment);
    synchronized (publicKeys) {
        implRegister(k);  // Selector注册该SelectionKey
    }
    k.interestOps(ops);  // 向该SelectionKey注册事件
    return k;
}

//  EPollArrayWrapper.java
void add(int fd) {
    // force the initial update events to 0 as it may be KILLED by a
    // previous registration.
    synchronized (updateLock) {
        assert !registered.get(fd);
        setUpdateEvents(fd, (byte)0, true);
    }
}

private void setUpdateEvents(int fd, byte events, boolean force) {
    if (fd < MAX_UPDATE_ARRAY_SIZE) {
        if ((eventsLow[fd] != KILLED) || force) {
            eventsLow[fd] = events;
        }
    } else {
        Integer key = Integer.valueOf(fd);
        if (!isEventsHighKilled(key) || force) {
            eventsHigh.put(key, Byte.valueOf(events));
        }
    }
}

private final byte[] eventsLow = new byte[MAX_UPDATE_ARRAY_SIZE];
private Map<Integer,Byte> eventsHigh;

void setInterest(int fd, int mask) {
    synchronized (updateLock) {
        // 如果updateDescriptors满了就扩容
        int oldCapacity = updateDescriptors.length;
        if (updateCount == oldCapacity) {
            int newCapacity = oldCapacity + INITIAL_PENDING_UPDATE_SIZE;
            int[] newDescriptors = new int[newCapacity];
            System.arraycopy(updateDescriptors, 0, newDescriptors, 0, oldCapacity);
            updateDescriptors = newDescriptors;
        }
        updateDescriptors[updateCount++] = fd;

        // 将要监听的fd上的事件存储起来
        byte b = (byte)mask;
        assert (b == mask) && (b != KILLED);
        setUpdateEvents(fd, b, false);
    }
}

// SelectorImpl.java
protected int doSelect(long timeout) throws IOException {
    if (closed)
        throw new ClosedSelectorException();
    processDeregisterQueue();
    try {
        begin();
        pollWrapper.poll(timeout);  // 关键
    } finally {
        end();
    }
    processDeregisterQueue();
    int numKeysUpdated = updateSelectedKeys();  // 关键
    if (pollWrapper.interrupted()) {
        // Clear the wakeup pipe
        pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
        synchronized (interruptLock) {
            pollWrapper.clearInterrupted();
            IOUtil.drain(fd0);
            interruptTriggered = false;
        }
    }
    return numKeysUpdated;
}

int poll(long timeout) throws IOException {
        updateRegistrations();  // 调用epollCtl更新事件监听
        updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);  // native函数
        for (int i=0; i<updated; i++) {
            if (getDescriptor(i) == incomingInterruptFD) {
                interruptedIndex = i;
                interrupted = true;
                break;
            }
        }
        return updated;
    }

/**
 * Update the keys whose fd's have been selected by the epoll.
 * Add the ready keys to the ready queue.
 * 更新selectedKeys以便之后this.selector.selectedKeys();使用
 */
private int updateSelectedKeys() {
    int entries = pollWrapper.updated;  // 之前native方法epollWait的返回值
    int numKeysUpdated = 0;
    for (int i=0; i<entries; i++) {
        int nextFD = pollWrapper.getDescriptor(i);  // pollWrapper是native数组, 也在epollWait中得到了更新
        SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
        // ski is null in the case of an interrupt
        if (ski != null) {
            int rOps = pollWrapper.getEventOps(i);
            if (selectedKeys.contains(ski)) {
                // 已经有该SelectionKey就更新事件
                if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
                    numKeysUpdated++;
                }
            } else {
                // 没有该SelectionKey则将其添加到selectedKeys
                ski.channel.translateAndSetReadyOps(rOps, ski);
                if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
                    selectedKeys.add(ski);
                    numKeysUpdated++;
                }
            }
        }
    }
    return numKeysUpdated;
}