netty源码阅读之NioEventLoop之NioEventLoop执行-----processSelectedKey()执行

川长思鸟来 2022-05-17 10:43 154阅读 0赞

从《[netty源码阅读之NioEventLoop之NioEventLoop执行][netty_NioEventLoop_NioEventLoop]》知道，select之后，就是processSelectedKey()

这里我们要学习两点：

1、selected keySet的优化

2、processSelectedKeysOptimized()执行逻辑

### 1、selected keySet的优化 ###

在《[netty源码阅读之NioEventLoop之NioEventLoop创建][netty_NioEventLoop_NioEventLoop 1]》这边文章中，我们在第二步中创建了一个selector，然后我们跟进去看看创建的过程，从selector = openSelector()进入：

private Selector openSelector() {
            final Selector selector;
            try {
                selector = provider.openSelector();
            } catch (IOException e) {
                throw new ChannelException("failed to open a new selector", e);
            }
    
            if (DISABLE_KEYSET_OPTIMIZATION) {
                return selector;
            }
    
            final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
    
            Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
                @Override
                public Object run() {
                    try {
                        return Class.forName(
                                "sun.nio.ch.SelectorImpl",
                                false,
                                PlatformDependent.getSystemClassLoader());
                    } catch (ClassNotFoundException e) {
                        return e;
                    } catch (SecurityException e) {
                        return e;
                    }
                }
            });
    
            if (!(maybeSelectorImplClass instanceof Class) ||
                    // ensure the current selector implementation is what we can instrument.
                    !((Class<?>) maybeSelectorImplClass).isAssignableFrom(selector.getClass())) {
                if (maybeSelectorImplClass instanceof Exception) {
                    Exception e = (Exception) maybeSelectorImplClass;
                    logger.trace("failed to instrument a special java.util.Set into: {}", selector, e);
                }
                return selector;
            }
    
            final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
    
            Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
                @Override
                public Object run() {
                    try {
                        Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
                        Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
    
                        selectedKeysField.setAccessible(true);
                        publicSelectedKeysField.setAccessible(true);
    
                        selectedKeysField.set(selector, selectedKeySet);
                        publicSelectedKeysField.set(selector, selectedKeySet);
                        return null;
                    } catch (NoSuchFieldException e) {
                        return e;
                    } catch (IllegalAccessException e) {
                        return e;
                    } catch (RuntimeException e) {
                        // JDK 9 can throw an inaccessible object exception here; since Netty compiles
                        // against JDK 7 and this exception was only added in JDK 9, we have to weakly
                        // check the type
                        if ("java.lang.reflect.InaccessibleObjectException".equals(e.getClass().getName())) {
                            return e;
                        } else {
                            throw e;
                        }
                    }
                }
            });
    
            if (maybeException instanceof Exception) {
                selectedKeys = null;
                Exception e = (Exception) maybeException;
                logger.trace("failed to instrument a special java.util.Set into: {}", selector, e);
            } else {
                selectedKeys = selectedKeySet;
                logger.trace("instrumented a special java.util.Set into: {}", selector);
            }
    
            return selector;
        }

首先用一句话来描述就是：把selector的keySet通过反射的方式，由原来HashSet的数据结构修改为数组的形式，使得原来在add操作的时候，可能的O(n)时间复杂度降低到O(1)。

下面我们分析：

selector = provider.openSelector();没什么好说，就是调用jdk底层，创建selector。

final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();创建一个keyset，这个keyset是由数组实现的，查看这个类的源码：

final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {
    
        private SelectionKey[] keysA;
        private int keysASize;
        private SelectionKey[] keysB;
        private int keysBSize;
        private boolean isA = true;
    
        SelectedSelectionKeySet() {
            keysA = new SelectionKey[1024];
            keysB = keysA.clone();
        }
    
        @Override
        public boolean add(SelectionKey o) {
            if (o == null) {
                return false;
            }
    
            if (isA) {
                int size = keysASize;
                keysA[size ++] = o;
                keysASize = size;
                if (size == keysA.length) {
                    doubleCapacityA();
                }
            } else {
                ...
            }
    
            return true;
        }
    
        private void doubleCapacityA() {
            ...
        }
         ...
        SelectionKey[] flip() {
            if (isA) {
                isA = false;
                keysA[keysASize] = null;
                keysBSize = 0;
                return keysA;
            } else {
                isA = true;
                keysB[keysBSize] = null;
                keysASize = 0;
                return keysB;
            }
        }
    
        @Override
        public int size() {
            if (isA) {
                return keysASize;
            } else {
                return keysBSize;
            }
        }
    
        @Override
        public boolean remove(Object o) {
            return false;
        }
    
        @Override
        public boolean contains(Object o) {
            return false;
        }
    
        @Override
        public Iterator<SelectionKey> iterator() {
            throw new UnsupportedOperationException();
        }
    }

由于我们的keyset不需要remove，contains，iterator等这些操作，所以可以替换为数组。另外，由于原来的keyset也是继承自AbstractSet，这个也是继承自AbstractSet，所以有替换的可能性。

回到原来的代码

Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
                @Override
                public Object run() {
                    try {
                        return Class.forName(
                                "sun.nio.ch.SelectorImpl",
                                false,
                                PlatformDependent.getSystemClassLoader());
                    } catch (ClassNotFoundException e) {
                        return e;
                    } catch (SecurityException e) {
                        return e;
                    }
                }
            });
    
            if (!(maybeSelectorImplClass instanceof Class) ||
                    // ensure the current selector implementation is what we can instrument.
                    !((Class<?>) maybeSelectorImplClass).isAssignableFrom(selector.getClass())) {
                if (maybeSelectorImplClass instanceof Exception) {
                    Exception e = (Exception) maybeSelectorImplClass;
                    logger.trace("failed to instrument a special java.util.Set into: {}", selector, e);
                }
                return selector;
            }

这里就是通过反射的方式获取selector的类，并且判断类型，查看SelectorImpl的源码，它的keyset确实是HashSet的实现：

protected Set<SelectionKey> selectedKeys = new HashSet();
        protected HashSet<SelectionKey> keys = new HashSet();
        private Set<SelectionKey> publicKeys;
        private Set<SelectionKey> publicSelectedKeys;
    
        protected SelectorImpl(SelectorProvider var1) {
            super(var1);
            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);
            }
    
        }

HashSet底层是HashMap实现的，由于存在hash冲突，hashMap添加元素的时候，最坏的情况下，时间复杂度为O(n)，而数组每次都添加都后面，时间复杂度都为O(1)。

下面的代码就是标准的反射的流程：

Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
                        Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
    
                        selectedKeysField.setAccessible(true);
                        publicSelectedKeysField.setAccessible(true);
    
                        selectedKeysField.set(selector, selectedKeySet);
                        publicSelectedKeysField.set(selector, selectedKeySet);

把selector类的域读出来，并把selector的特定的域也就是keyset替换为新的提高性能的keyset。

上面的流程中，如果修改不成功，就返回原来没优化的selector

2、processSelectedKeysOptimized()执行逻辑

在NioEventLoop里面点进去processSelectedKeys()，查找它优化过的实现：

private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
            for (int i = 0;; i ++) {
                final SelectionKey k = selectedKeys[i];
                if (k == null) {
                    break;
                }
                // null out entry in the array to allow to have it GC'ed once the Channel close
                // See https://github.com/netty/netty/issues/2363
                selectedKeys[i] = null;
    
                final Object a = k.attachment();
    
                if (a instanceof AbstractNioChannel) {
                    processSelectedKey(k, (AbstractNioChannel) a);
                } else {
                    @SuppressWarnings("unchecked")
                    NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                    processSelectedKey(k, task);
                }
    
                if (needsToSelectAgain) {
                    // null out entries in the array to allow to have it GC'ed once the Channel close
                    // See https://github.com/netty/netty/issues/2363
                    for (;;) {
                        i++;
                        if (selectedKeys[i] == null) {
                            break;
                        }
                        selectedKeys[i] = null;
                    }
    
                    selectAgain();
                    // Need to flip the optimized selectedKeys to get the right reference to the array
                    // and reset the index to -1 which will then set to 0 on the for loop
                    // to start over again.
                    //
                    // See https://github.com/netty/netty/issues/1523
                    selectedKeys = this.selectedKeys.flip();
                    i = -1;
                }
            }
        }

在这里，这个k.attachment就是NioServerSocketChannel，我们在之前有一篇文章说过，会把netty包装的channel放进去，就是这里去处理。判断到是AbstractNioChannel的实例，就会进行里面的操作：

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
            final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
            if (!k.isValid()) {
                final EventLoop eventLoop;
                try {
                    eventLoop = ch.eventLoop();
                } catch (Throwable ignored) {
                    // If the channel implementation throws an exception because there is no event loop, we ignore this
                    // because we are only trying to determine if ch is registered to this event loop and thus has authority
                    // to close ch.
                    return;
                }
                // Only close ch if ch is still registerd to this EventLoop. ch could have deregistered from the event loop
                // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
                // still healthy and should not be closed.
                // See https://github.com/netty/netty/issues/5125
                if (eventLoop != this || eventLoop == null) {
                    return;
                }
                // close the channel if the key is not valid anymore
                unsafe.close(unsafe.voidPromise());
                return;
            }
    
            try {
                int readyOps = k.readyOps();
                // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
                // the NIO JDK channel implementation may throw a NotYetConnectedException.
                if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                    // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                    // See https://github.com/netty/netty/issues/924
                    int ops = k.interestOps();
                    ops &= ~SelectionKey.OP_CONNECT;
                    k.interestOps(ops);
    
                    unsafe.finishConnect();
                }
    
                // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
                if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                    // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                    ch.unsafe().forceFlush();
                }
    
                // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
                // to a spin loop
                if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                    unsafe.read();
                    if (!ch.isOpen()) {
                        // Connection already closed - no need to handle write.
                        return;
                    }
                }
            } catch (CancelledKeyException ignored) {
                unsafe.close(unsafe.voidPromise());
            }
        }

轮询各种各样的IO事件，例如OP\_READ和OP\_ACCEPT等。这些IO操作，都是通过底层的unsafe去实现的。

[netty_NioEventLoop_NioEventLoop]: https://blog.csdn.net/fst438060684/article/details/81558585
[netty_NioEventLoop_NioEventLoop 1]: https://blog.csdn.net/fst438060684/article/details/81534359