【335期】面试官：你了解Netty事件注册过程吗

Netty是对NIO的封装，通过事件驱动的网络编程框架，自然是要实现NIO中的事件注册与监听。在NIO中我们都是显式的注册每一个事件，但是Netty为开发人员封装了这些细节，提供了简单易用的API，底层是如何实现的呢，这就是本篇文章要讨论的问题。

NIO的SelectionKey中有四种事件，可读、可写、连接、接收连接

//SingleThreadEventExecutor
private void doStartThread() {
    assert thread == null;
    executor.execute(new Runnable() {
    @Override
    public void run() {
       thread = Thread.currentThread();
       if (interrupted) {
           thread.interrupt();
       }

       boolean success = false;
       updateLastExecutionTime();
       try {
           SingleThreadEventExecutor.this.run();
           success = true;
       } catch (Throwable t) {
           logger.warn("Unexpected exception from an event executor: ", t);
       }
      }
     }
    }

protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
                    // fallthrough
            }

            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                processSelectedKeys();
                runAllTasks();
            } else {
                final long ioStartTime = System.nanoTime();

                processSelectedKeys();

                final long ioTime = System.nanoTime() - ioStartTime;
                runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
            }

            if (isShuttingDown()) {
                closeAll();
                if (confirmShutdown()) {
                    break;
                }
            }
        } catch (Throwable t) {
            logger.warn("Unexpected exception in the selector loop.", t);

            // Prevent possible consecutive immediate failures that lead to
            // excessive CPU consumption.
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // Ignore.
            }
        }
    }
}

private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized(selectedKeys.flip());
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
    for (int i = 0;; i ++) {
        final SelectionKey k = selectedKeys[i];
        if (k == null) {
            break;
        }
        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);
        }
    }
}
// NioEventLoop处理事件分发 
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    try {
        int readyOps = k.readyOps();
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
            if (!ch.isOpen()) {
                return;
            }
        }
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            ch.unsafe().forceFlush();
        }
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

//AbstractNioMessageChannel
   public void read() {
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        final ChannelPipeline pipeline = pipeline();
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.reset(config);

        boolean closed = false;
        Throwable exception = null;
        try {
            try {
                do {
                    int localRead = doReadMessages(readBuf);
                    if (localRead == 0) {
                        break;
                    }
                    if (localRead < 0) {
                        closed = true;
                        break;
                    }

                    allocHandle.incMessagesRead(localRead);
                } while (allocHandle.continueReading());
            } catch (Throwable t) {
                exception = t;
            }

            int size = readBuf.size();
            for (int i = 0; i < size; i ++) {
                readPending = false;
                pipeline.fireChannelRead(readBuf.get(i));
            }
            readBuf.clear();
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();

            if (exception != null) {
                if (exception instanceof IOException && !(exception instanceof PortUnreachableException)) {
                    closed = !(AbstractNioMessageChannel.this instanceof ServerChannel);
                }
                pipeline.fireExceptionCaught(exception);
            }

            if (closed) {
                inputShutdown = true;
                if (isOpen()) {
                    close(voidPromise());
                }
            }
        } finally {
            if (!readPending && !config.isAutoRead()) {
                removeReadOp();
            }
        }
    }
}
NioServerSocketChannel.doReadMessage方法  
protected int doReadMessages(List<Object> buf) throws Exception {
   SocketChannel ch = javaChannel().accept();

   try {
       if (ch != null) {
           buf.add(new NioSocketChannel(this, ch));
           return 1;
       }
   } catch (Throwable t) {
   return 0;
}

public NioSocketChannel(Channel parent, EventLoop eventLoop, SocketChannel socket) {  
    super(parent, eventLoop, socket);  
    config = new DefaultSocketChannelConfig(this, socket.socket());  
}  

protected AbstractNioByteChannel(Channel parent, EventLoop eventLoop, SelectableChannel ch) {  
    super(parent, eventLoop, ch, SelectionKey.OP_READ);  
}  

protected AbstractNioChannel(Channel parent, EventLoop eventLoop, SelectableChannel ch, int readInterestOp) {  
   super(parent, eventLoop);  
   this.ch = ch;  
   this.readInterestOp = readInterestOp;  
   try {  
       ch.configureBlocking(false);  
   } catch (IOException e) {  
       try {  
           ch.close();  
       } catch (IOException e2) {
       } 
   }  
}  

// ServerBootstrapAcceptor.channelRead方法来注册Channel  
public void channelRead(ChannelHandlerContext ctx, Object msg) {  
    Channel child = (Channel) msg;  

    child.pipeline().addLast(childHandler);  

    for (Entry<ChannelOption<?>, Object> e: childOptions) {  
        try {  
            if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {  
                logger.warn("Unknown channel option: " + e);  
            }  
        } catch (Throwable t) {  
            logger.warn("Failed to set a channel option: " + child, t);  
        }  
    }  

    for (Entry<AttributeKey<?>, Object> e: childAttrs) {  
        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());  
    }  

    child.unsafe().register(child.newPromise());  
}  

// AbstractChannel的register0方法，先注册，后fireChannelActive来调用HeadHandler设置SelectionKey的interestOps  
private void register0(ChannelPromise promise) {  
   try {  
       // check if the channel is still open as it could be closed in the mean time when the register  
       // call was outside of the eventLoop  
       if (!ensureOpen(promise)) {  
           return;  
       }  
       doRegister();  
       registered = true;  
       promise.setSuccess();  
       pipeline.fireChannelRegistered();  
       if (isActive()) {  
           pipeline.fireChannelActive();  
       }  
   } catch (Throwable t) {  
   }  
}  



// AbstractNioChannel 注册Channel到selector,只注册空事件，具体OP_READ事件延迟注册  
protected void doRegister() throws Exception {  
    boolean selected = false;  
    for (;;) {  
        try {  
            selectionKey = javaChannel().register(eventLoop().selector, 0, this);  
            return;  
        } catch (CancelledKeyException e) {  
            if (!selected) { 
                eventLoop().selectNow();  
                selected = true;  
            } else {  
              throw e;  
            }  
        }  
    }  
}  


 protected void doBeginRead() throws Exception {  
    if (inputShutdown) {  
        return;  
    }  

    final SelectionKey selectionKey = this.selectionKey;  
    if (!selectionKey.isValid()) {  
        return;  
    }  
// interestOps = 0, readInterestOps = OP_READ = 1  
    final int interestOps = selectionKey.interestOps();  
    if ((interestOps & readInterestOp) == 0) {  
        selectionKey.interestOps(interestOps | readInterestOp);  
    }  
}